Loading...
Searching...
No Matches
stream_expand_support.cpp
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.h
// Author: Aous Naman
// Date: 18 April 2024
//***************************************************************************/
#include <cassert>
#include <cstddef>
#include "ojph_threads.h"
#include "threaded_frame_processors.h"
#include "stream_expand_support.h"
namespace ojph
{
namespace stex
{
//
//
// static comparison functions
//
//
// Compares two 32 bit values, A with B, with the possibility A or B has
// undergone overflow. This problem has no proper solution, but here we
// assume that the value B approximately divides the space into two regions,
// a region larger than B and a region smaller than B. This leaves one
// undetermined value that lies at the opposite end of B, a case we just
// ignore -- it is part of smaller.
// NB: This is my current thinking -- I might be wrong
static inline bool is_greater32(ui32 a, ui32 b)
{
ui32 c = a - b;
return (c > 0u && c <= 0x7FFFFFFFu);
}
// Compares two 32 bit values, A with B, with the possibility A or B has
// undergone overflow. This problem has no proper solution, but here we
// assume that the value B approximately divides the space into two regions,
// a region larger than B and a region smaller than B. This leaves one
// undetermined value that lies at the opposite end of B, a case we just
// ignore -- it is part of smaller.
// NB: This is my current thinking -- I might be wrong
static inline bool is_smaller32(ui32 a, ui32 b)
{
ui32 c = a - b;
return (c >= 0x80000000u && c <= 0xFFFFFFFFu);
}
static inline bool is_greater24(ui32 a, ui32 b)
{ return is_greater32(a << 8, b << 8); }
static inline bool is_smaller24(ui32 a, ui32 b)
{ return is_smaller32(a << 8, b << 8); }
static inline ui32 clip_seq_num(ui32 n) { return (n & 0xFFFFFF); }
//
//
//
//
//
void packets_handler::init(bool quiet, ui32 num_packets,
frames_handler* frames)
{
assert(this->num_packets == 0);
avail = packet_store = new rtp_packet[num_packets];
ui32 i = 0;
for (; i < num_packets - 1; ++i)
packet_store[i].init(packet_store + i + 1);
packet_store[i].init(NULL);
this->quiet = quiet;
this->num_packets = num_packets;
this->frames = frames;
}
rtp_packet* packets_handler::exchange(rtp_packet* p)
{
assert(num_packets > 0 && p == in_use);
if (p != NULL) {
if (p->num_bytes == 0)
return p;
if (last_seq_num == 0) // initialization
last_seq_num = clip_seq_num(p->get_seq_num() - 1);
// packet is old, and is ignored -- no need to included it in the
// lost packets, because this packet was considered lost previously.
// This also captures the case where the previous packet and this packet
// has the same sequence number, which is rather weird but possible
// if some intermediate network unit retransmits packets.
if (is_smaller24(p->get_seq_num(), clip_seq_num(last_seq_num + 1)))
return p;
else if (p->get_seq_num() == clip_seq_num(last_seq_num + 1))
{
consume_packet();
// see if we can push one packet from the top of the buffer
if (in_use && in_use->get_seq_num() == clip_seq_num(last_seq_num + 1))
consume_packet();
}
else // sequence larger than expected
{
// Place the packet in the in_use queue according to its sequence
// number; we may have to move it down the queue. The in_use queue is
// always arranged in an ascending order, where the top of the queue
// (pointed to by in_use) has the smallest sequence number.
if (in_use->next != NULL) // we have more than 1 packet in queue
{
rtp_packet* t = in_use;
while (t->next != NULL &&
is_greater24(p->get_seq_num(), t->next->get_seq_num()))
t = t->next;
if (t->next != NULL && p->get_seq_num() == t->next->get_seq_num())
{ // this is a repeated packet and must be removed
in_use = in_use->next;
p->next = avail;
avail = p;
}
else {
if (t == in_use) // at front of queue -- exactly where it should be
{ } // do nothing
else if (t->next == NULL) { // at the end of queue
in_use = in_use->next; // remove p from the queue
t->next = p;
p->next = NULL;
}
else { // in the middle of the queue
in_use = in_use->next; // p removed from the start of queue
p->next = t->next;
t->next = p;
}
}
}
// If avail == NULL, all packets are being used (in_use), meaning
// the queue is already full. We push packets from to the top of in_use
// queue.
// If avail != NULL, we push one packet from the top of the buffer,
// if it has the correct sequence number.
if (avail == NULL ||
in_use->get_seq_num() == clip_seq_num(last_seq_num + 1))
{
if (avail == NULL)
lost_packets +=
in_use->get_seq_num() - clip_seq_num(last_seq_num + 1);
consume_packet();
if (in_use && in_use->get_seq_num() == clip_seq_num(last_seq_num + 1))
consume_packet();
}
}
}
// move from avail to in_use -- there must be at least one packet in avail
assert(avail != NULL);
p = avail;
avail = avail->next;
p->next = in_use;
in_use = p;
return p;
}
void packets_handler::flush()
{
// move all packets from in_use to avail
while (in_use)
{
rtp_packet *p = in_use;
in_use = in_use->next;
p->next = avail;
avail = p;
}
}
void packets_handler::consume_packet()
{
last_seq_num = in_use->get_seq_num();
frames->push(in_use);
// move pack from in_use to avail; the packet must be equal to in_use
rtp_packet* p = in_use;
in_use = in_use->next;
p->next = avail;
avail = p;
}
//
//
//
//
//
void stex_file::notify_file_completion()
{
int t = done.fetch_add(-1, std::memory_order_acq_rel);
if (t == 1) // done is 0
parent->increment_num_complete_files();
}
//
//
//
//
//
frames_handler::~frames_handler()
{
if (storers_store)
delete[] storers_store;
if (files_store)
delete[] files_store;
}
void frames_handler::init(bool quiet, const char *target_name,
thds::thread_pool* thread_pool)
{
this->quiet = quiet;
this->num_threads = (ui32)thread_pool->get_num_threads();
this->target_name = target_name;
num_files = num_threads + 1;
avail = files_store = new stex_file[num_files];
storers_store = new j2k_frame_storer[num_files];
ui32 i = 0;
for (; i < num_files - 1; ++i) {
files_store[i].f.open(2 << 20, false);
files_store[i].f.close();
files_store[i].init(this, files_store + i + 1, storers_store + i,
target_name);
storers_store[i].init(files_store + i, target_name);
}
files_store[i].f.open(2 << 20, false);
files_store[i].f.close();
files_store[i].init(this, NULL, storers_store + i, target_name);
storers_store[i].init(files_store + i, target_name);
this->thread_pool = thread_pool;
}
void frames_handler::push(rtp_packet* p)
{
assert(!is_smaller32(p->get_time_stamp(), last_time_stamp));
assert(!is_smaller24(p->get_seq_num(), last_seq_number));
last_seq_number = p->get_seq_num();
// check if any of the frames processed in other threads are done
check_files_in_processing();
// process newly received packet
if (p->get_packet_type() != rtp_packet::PT_BODY)
{ // main packet payload
// The existence of a previous frame means we did not get the marked
// packet. Here, we close the frame and move it to processing
if (in_use) {
++trunc_frames;
send_to_processing();
}
// This is where we process a new frame, if there is space
if (avail)
{
// move from avail to in_use
in_use = avail;
avail = avail->next;
in_use->next = NULL;
assert(in_use->done.load(std::memory_order_acquire) == 0);
in_use->time_stamp = p->get_time_stamp();
in_use->last_seen_seq = p->get_seq_num();
in_use->frame_idx = total_frames;
in_use->f.open();
in_use->f.write(p->get_data(), p->get_data_size());
}
else
++lost_frames;
++total_frames;
last_time_stamp = p->get_time_stamp();
}
else
{ // body packet payload
if (in_use != NULL)
{
if (p->get_time_stamp() == in_use->time_stamp)
{ // this is a continuation of a previous frame
if (p->get_seq_num() == clip_seq_num(in_use->last_seen_seq + 1))
{
in_use->last_seen_seq = p->get_seq_num();
in_use->f.write(p->get_data(), p->get_data_size());
if (p->is_marked())
send_to_processing();
}
else {
// we must have missed packets
++trunc_frames;
send_to_processing();
}
}
else
{
// This is a different frame and we did not get the marked packet.
// We close the older frame and send it for processing
++trunc_frames;
send_to_processing();
if (is_greater32(p->get_time_stamp(), last_time_stamp))
{
++total_frames;
last_time_stamp = p->get_time_stamp();
}
}
}
else // no frame is being written
{
if (is_greater32(p->get_time_stamp(), last_time_stamp))
{
++total_frames;
last_time_stamp = p->get_time_stamp();
}
}
}
}
void frames_handler::get_stats(ui32& total_frames, ui32& trunc_frames,
ui32& lost_frames)
{
total_frames = this->total_frames;
trunc_frames = this->trunc_frames;
lost_frames = this->lost_frames;
}
bool frames_handler::flush()
{
// check if any of the frames processed in other threads are done
check_files_in_processing();
// check the file in in_use and terminate it
if (in_use != NULL)
{
// move from in_use to avail
in_use->f.close();
in_use->next = avail;
avail = in_use;
in_use = NULL;
}
return (processing != NULL);
}
void frames_handler::check_files_in_processing()
{
// check if any of the frames processed in other threads are done
int nf = num_complete_files.load(std::memory_order_acquire);
if (nf > 0)
{
stex_file* f = processing, *pf = NULL;
while(f != NULL && nf > 0)
{
num_complete_files.fetch_add(-1, std::memory_order_relaxed);
if (f->done.load(std::memory_order_acquire) == 0)
{
// move f from processing to avail
f->time_stamp = 0;
f->last_seen_seq = 0;
f->frame_idx = 0;
if (f == processing)
{
processing = processing->next;
f->next = avail;
avail = f;
f = processing; // for next test
}
else {
pf->next = f->next;
f->next = avail;
avail = f;
f = pf->next; // for next test
}
}
else
{
pf = f;
f = f->next;
}
nf = num_complete_files.load(std::memory_order_acquire);
}
}
}
void frames_handler::send_to_processing()
{
in_use->f.close();
if (target_name) {
in_use->next = processing;
processing = in_use;
in_use->done.store(1, std::memory_order_relaxed);
thread_pool->add_task(in_use->storer);
}
else {
in_use->next = avail;
avail = in_use;
}
in_use = NULL;
}
} // !stex namespace
} // !ojph namespace
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::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::~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::packet_store
rtp_packet * packet_store
address of packet memory allocation
Definition stream_expand_support.h:327
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::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::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::stex::is_smaller32
static bool is_smaller32(ui32 a, ui32 b)
Definition stream_expand_support.cpp:79
ojph::stex::is_smaller24
static bool is_smaller24(ui32 a, ui32 b)
Definition stream_expand_support.cpp:90
ojph::stex::is_greater32
static bool is_greater32(ui32 a, ui32 b)
Definition stream_expand_support.cpp:65
ojph::stex::is_greater24
static bool is_greater24(ui32 a, ui32 b)
Definition stream_expand_support.cpp:86
ojph::stex::clip_seq_num
static ui32 clip_seq_num(ui32 n)
Definition stream_expand_support.cpp:94
ojph
Definition ojph_img_io.h:52
ojph::ui32
uint32_t ui32
Definition ojph_defs.h:54
ojph_threads.h
stream_expand_support.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_time_stamp
ui32 get_time_stamp()
Definition stream_expand_support.h:117
ojph::stex::rtp_packet::get_seq_num
ui32 get_seq_num()
Definition stream_expand_support.h:112
ojph::stex::rtp_packet::get_data_size
ui32 get_data_size()
Definition stream_expand_support.h:138
ojph::stex::rtp_packet::get_data
ui8 * get_data()
Definition stream_expand_support.h:136
ojph::stex::rtp_packet::next
rtp_packet * next
used for linking packets
Definition stream_expand_support.h:210
ojph::stex::rtp_packet::get_packet_type
ui32 get_packet_type()
Definition stream_expand_support.h:123
ojph::stex::rtp_packet::PT_BODY
@ PT_BODY
Definition stream_expand_support.h:86
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::frame_idx
ui32 frame_idx
frame number in the sequence
Definition stream_expand_support.h:405
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::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
threaded_frame_processors.h