<commit_before>

#include "samson/common/coding.h"           // KVFILE_NUM_HASHGROUPS
#include "samson/common/SamsonSetup.h"      // samson::SamsonSetup


#include "StreamManager.h"          // samson::stream::StreamManager
#include "BlockBreakQueueTask.h"    // samson::stream::BlockBreakQueueTask
#include "Block.h"                  // samson::stream::Block
#include "BlockList.h"              // samson::stream::BlockList
#include "Queue.h"                  // OwnInterface


size_t next_pow_2( size_t value )
{
    if( value < 2)
        return 1;

    int p = 1;
    while ( true )
    {
	   if( value == pow(2.0 , p)) 
           return pow( 2.0 , p);
        
        if( value < pow( 2.0 , p ) )
            return pow( 2.0 , p );
        p++;
    }
    
    LM_X(1,("Internal error"));
    return 1;
    
}

namespace samson {
    namespace stream
    {
        
        
        
        void BlockIdList::addIds( BlockList *list )
        {
            au::list< Block >::iterator block_it;
            for ( block_it = list->blocks.begin() ; block_it != list->blocks.end() ; block_it++)
            {
                Block* block = *block_it;
                size_t block_id = block->getId();
                addId( block_id );
            }
            
        }
        void BlockIdList::removeIds( BlockList *list )
        {
            au::list< Block >::iterator block_it;
            for ( block_it = list->blocks.begin() ; block_it != list->blocks.end() ; block_it++)
            {
                Block* block = *block_it;
                size_t block_id = block->getId();
                removeId( block_id );
            }
        }
        
        void BlockIdList::addId( size_t id )
        {
            block_ids.insert( id );
        }
        void BlockIdList::removeId( size_t id )
        {
            block_ids.erase( id );
        }
        
        size_t BlockIdList::num_ids()
        {
            return block_ids.size();
        }
        
        bool BlockIdList::containsBlockId( size_t id )
        {
            return ( block_ids.find( id ) != block_ids.end() );
        }
        
        
#pragma mark Queue
        
        Queue::Queue( std::string _name , StreamManager* _streamManager  )
        {
            // Keep the name
            name = _name;

            // Still pending of the block
            format = KVFormat("*","*");
            
            // Create a list for blocks pending to be broken...
            list = new BlockList( "queue_" + name );
            
            // Pointer to StreamManager
            streamManager = _streamManager;
            
            // By default it is not paused
            paused = false;
            
            // Number of divisions for this queue
            num_divisions = 1;
            
            // Number of updates ( 0 at start )
            updates = 0;
        }
        
        Queue::~Queue()
        {
            delete list;
        }

        
        void Queue::push( BlockList *list )
        {
            au::list< Block >::iterator b;
            for (b = list->blocks.begin() ; b != list->blocks.end() ; b++ )
                push( *b );
        }
        
        void Queue::push( Block *block )
        {
            
            if( format == KVFormat("*","*") )
                format = block->header->getKVFormat();  // Get the format of the first one...
            else
            {
                if( format != block->header->getKVFormat() )
                {
                    LM_W(("Trying to push a block with format %s to queue %s with format %s. Block rejected", block->header->getKVFormat().str().c_str() , name.c_str() , format.str().c_str() ));
                    return;
                }
            }
            
            // Add to the list for this queue
            list->add( block );
            
            // Review this queue
            review();
        }        
        
        void Queue::getInfo( std::ostringstream& output)
        {
            au::xml_open(output, "queue");

            au::xml_simple(output, "name", name );

            au::xml_simple(output, "num_divisions", num_divisions);
            
            if( paused )
                au::xml_simple(output, "paused", "YES" );
            else
                au::xml_simple(output, "paused", "NO" );

            // Information about the format
            format.getInfo(output);

            // Block list information
            list->getInfo( output );
            
            // Informat description of queue status
            std::ostringstream status;
            
            if( updating_divisions.size() > 0)
            {
                status << au::str("[ Updating state %d / %d ] " , updating_divisions.size() , num_divisions );
                
                if( breaking_block_ids.num_ids() > 0)
                    status << "[ Error: breaking tasks while updating state ] ";
            }
            
            if( getMinNumDivisions() > num_divisions )
            {
                status << au::str("[ Moving from %d to %d divisions ] " , num_divisions , getMinNumDivisions() );
            }
            
            if( breaking_block_ids.num_ids() > 0)
                status << au::str("[ Breaking %d/%d blocks ] " , breaking_block_ids.num_ids() , list->getNumBlocks()); 

            if( processing_block_ids.num_ids() > 0)
                status << au::str("[ Processing %d/%d blocks ] " , processing_block_ids.num_ids() , list->getNumBlocks()); 
            
            if( status.str() == "" )
                status << "ready";
                                 
            au::xml_simple( output , "status" , status.str() );

            // Information about content
            BlockInfo block_info;
            update( block_info );
            block_info.getInfo(output);
            
            au::xml_close(output, "queue");
        }
        
        void Queue::update( BlockInfo& block_info )
        {
            list->update( block_info );
            
        }
        
        void Queue::setProperty( std::string property , std::string value )
        {
            environment.set( property , value );<commit_msg>Minor update for queue status report<commit_after>

#include "samson/common/coding.h"           // KVFILE_NUM_HASHGROUPS
#include "samson/common/SamsonSetup.h"      // samson::SamsonSetup


#include "StreamManager.h"          // samson::stream::StreamManager
#include "BlockBreakQueueTask.h"    // samson::stream::BlockBreakQueueTask
#include "Block.h"                  // samson::stream::Block
#include "BlockList.h"              // samson::stream::BlockList
#include "Queue.h"                  // OwnInterface


size_t next_pow_2( size_t value )
{
    if( value < 2)
        return 1;

    int p = 1;
    while ( true )
    {
	   if( value == pow(2.0 , p)) 
           return pow( 2.0 , p);
        
        if( value < pow( 2.0 , p ) )
            return pow( 2.0 , p );
        p++;
    }
    
    LM_X(1,("Internal error"));
    return 1;
    
}

namespace samson {
    namespace stream
    {
        
        
        
        void BlockIdList::addIds( BlockList *list )
        {
            au::list< Block >::iterator block_it;
            for ( block_it = list->blocks.begin() ; block_it != list->blocks.end() ; block_it++)
            {
                Block* block = *block_it;
                size_t block_id = block->getId();
                addId( block_id );
            }
            
        }
        void BlockIdList::removeIds( BlockList *list )
        {
            au::list< Block >::iterator block_it;
            for ( block_it = list->blocks.begin() ; block_it != list->blocks.end() ; block_it++)
            {
                Block* block = *block_it;
                size_t block_id = block->getId();
                removeId( block_id );
            }
        }
        
        void BlockIdList::addId( size_t id )
        {
            block_ids.insert( id );
        }
        void BlockIdList::removeId( size_t id )
        {
            block_ids.erase( id );
        }
        
        size_t BlockIdList::num_ids()
        {
            return block_ids.size();
        }
        
        bool BlockIdList::containsBlockId( size_t id )
        {
            return ( block_ids.find( id ) != block_ids.end() );
        }
        
        
#pragma mark Queue
        
        Queue::Queue( std::string _name , StreamManager* _streamManager  )
        {
            // Keep the name
            name = _name;

            // Still pending of the block
            format = KVFormat("*","*");
            
            // Create a list for blocks pending to be broken...
            list = new BlockList( "queue_" + name );
            
            // Pointer to StreamManager
            streamManager = _streamManager;
            
            // By default it is not paused
            paused = false;
            
            // Number of divisions for this queue
            num_divisions = 1;
            
            // Number of updates ( 0 at start )
            updates = 0;
        }
        
        Queue::~Queue()
        {
            delete list;
        }

        
        void Queue::push( BlockList *list )
        {
            au::list< Block >::iterator b;
            for (b = list->blocks.begin() ; b != list->blocks.end() ; b++ )
                push( *b );
        }
        
        void Queue::push( Block *block )
        {
            
            if( format == KVFormat("*","*") )
                format = block->header->getKVFormat();  // Get the format of the first one...
            else
            {
                if( format != block->header->getKVFormat() )
                {
                    LM_W(("Trying to push a block with format %s to queue %s with format %s. Block rejected", block->header->getKVFormat().str().c_str() , name.c_str() , format.str().c_str() ));
                    return;
                }
            }
            
            // Add to the list for this queue
            list->add( block );
            
            // Review this queue
            review();
        }        
        
        void Queue::getInfo( std::ostringstream& output)
        {
            au::xml_open(output, "queue");

            au::xml_simple(output, "name", name );

            au::xml_simple(output, "num_divisions", num_divisions);
            
            if( paused )
                au::xml_simple(output, "paused", "YES" );
            else
                au::xml_simple(output, "paused", "NO" );

            // Information about the format
            format.getInfo(output);

            // Block list information
            list->getInfo( output );
            
            // Informat description of queue status
            std::ostringstream status;
            
            if( updating_divisions.size() > 0)
            {
                status << au::str("[ Updating state %d / %d ] " , updating_divisions.size() , num_divisions );
                
                if( breaking_block_ids.num_ids() > 0)
                    status << "[ Error: breaking tasks while updating state ] ";
                
                if( getMinNumDivisions() > num_divisions )
                {
                    status << au::str("[ Moving from %d to %d divisions ] " , num_divisions , getMinNumDivisions() );
                }

            }
            
            
            if( breaking_block_ids.num_ids() > 0)
                status << au::str("[ Breaking %d/%d blocks ] " , breaking_block_ids.num_ids() , list->getNumBlocks()); 

            if( processing_block_ids.num_ids() > 0)
                status << au::str("[ Processing %d/%d blocks ] " , processing_block_ids.num_ids() , list->getNumBlocks()); 
            
            if( status.str() == "" )
                status << "ready";
                                 
            au::xml_simple( output , "status" , status.str() );

            // Information about content
            BlockInfo block_info;
            update( block_info );
            block_info.getInfo(output);
            
            au::xml_close(output, "queue");
        }
        
        void Queue::update( BlockInfo& block_info )
        {
            list->update( block_info );
            
        }
        
        void Queue::setProperty( std::string property , std::string value )
        {
            environment.set( property , value );
{'fraction_non_alphanumeric': 0.11701702501798418, 'fraction_numerical': 0.0033570457996962674, 'mean_word_length': 1.2560403894698882, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 57, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 14}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7292594', 'n_tokens_mistral': 3477, 'n_tokens_neox': 3376, 'n_words': 785}