<commit_before>#include <pthread.h>
#include <iostream>                 // std::cerr
#include <signal.h>                 // kill(.)
#include <stdlib.h>                 // exit()
#include <sys/types.h>      
#include <sys/wait.h>               // waitpid()

#include "logMsg/logMsg.h"                 // LM_*
#include "logMsg/traceLevels.h"            // LmtIsolated, etc. 

#include "au/gpb.h"                     // au::readGPB & au::writeGPB
#include "au/ErrorManager.h"

#include "engine/MemoryManager.h"			// engine::MemoryManager
#include "engine/Notification.h"            // engine::Notification

#include "samson/common/traces.h"                       // logFd
#include "samson/common/SamsonSetup.h"                // Goyo: ss:SamsonSetup
#include "samson/isolated/ProcessIsolated.h"         //Goyo: For defines WORKER_TASK_ITEM_CODE_FLUSH_BUFFER
#include "samson/common/NotificationMessages.h"     // notification_samson_worker_send_trace
#include "samson/common/MemoryTags.h"                 // MemoryInput , MemoryOutputNetwork, ...

#include "samson/network/iomMsgAwait.h"            // iomMsgAwait
#include "samson/network/Packet.h"                     // samson::Packet


#include "samson/isolated/ProcessItemIsolated.h"    // Own interface
#include "samson/isolated/SharedMemoryManager.h"    

#include "SharedMemoryItem.h"           // engine::SharedMemoryItem


namespace samson
{

   
    
    // Static variable to active the "thread" mode of the background process execution
    bool ProcessItemIsolated::isolated_process_as_tread = false;
    	
	
	// Function to run by the backgroudn thread
	void* run_ProcessItemIsolated( void* p )
	{
        // Free resources automatically when this thread finish
        pthread_detach(pthread_self());
        
		ProcessItemIsolated* tmp = (ProcessItemIsolated*)p;
		tmp->runBackgroundProcessRun();
		return NULL;
	}
	
	ProcessItemIsolated::ProcessItemIsolated( std::string _description ) : ProcessItem( PI_PRIORITY_NORMAL_OPERATION ) 	
	{
        processItemIsolated_description = _description;
	}	

	ProcessItemIsolated::~ProcessItemIsolated()
    {
    }
	
    
	void ProcessItemIsolated::run()
	{
        LM_T( LmtIsolated , ("Isolated process %s: start ******************************************* ",getStatus().c_str()));
        
        
        if( isolated_process_as_tread )
            LM_T( LmtIsolated , ("Isolated process %s start in thread mode",getStatus().c_str()));
        else
            LM_T( LmtIsolated , ("Isolated process %s start in fork mode",getStatus().c_str()));
        			
		// Create a couple of pipes to communicate both process
		if ( pipe(pipeFdPair1) != 0 )
		{
			LM_E(("System error: not possible to create pipes when running this process"));
			error.set("System error: not possible to create pipes when running this process");
			return;
		}
		if ( pipe(pipeFdPair2) != 0 )
		{
			LM_E(("System error: not possible to create pipes when running this process"));
			error.set("System error: not possible to create pipes when running this process");
			LM_T( LmtIsolated, ("Isolated process %s: pipes closed: pipeFdPair1[0]:%d, pipeFdPair1[1]:%d\n", pipeFdPair1[0], pipeFdPair1[1]));
			close( pipeFdPair1[0] );
			close( pipeFdPair1[1] );
			return;
		}
		
		LM_T(LmtIsolated, ("Isolated process %s: pipes created. pipeFdPair1[0]:%d, pipeFdPair1[1]:%d, pipeFdPair2[0]:%d, pipeFdPair2[1]:%d\n",getStatus().c_str(), pipeFdPair1[0], pipeFdPair1[1], pipeFdPair2[0], pipeFdPair2[1]));
		LM_T( LmtIsolated , ("Isolated process %s: pipes created ",getStatus().c_str()));
		
		// Create the other process to run the other side
        pid_t pid = 0;
        if( isolated_process_as_tread )
        {
            ProcessItemIsolated *tmp = (this);
            pthread_t t;
            pthread_create(&t, NULL, run_ProcessItemIsolated, tmp);
        }
        else
        {
        	LM_T( LmtIsolated , ("Isolated process %s: father about to fork",getStatus().c_str()));
            pid = fork();
            if ( pid < 0 )
                LM_X(1,("Fork return an error"));
            
            if( pid == 0 )	// Children running the background process
            {
                runBackgroundProcessRun();
                LM_T(LmtIsolated,("Child in Background process finished, calling _exit that will close its side pipes: pipeFdPair1[1]:%d, pipeFdPair2[0]:%d\n", pipeFdPair1[1], pipeFdPair2[0]));
                _exit(1000);
            }
        }
		
        // Exchange all the necessary messages between background and foreground process
        LM_T( LmtIsolated , ("Isolated process %s: father runExchangeMessages start, child pid=%d ",getStatus().c_str(), pid));
        
        runExchangeMessages();
        
        LM_T( LmtIsolated , ("Isolated process %s: father runExchangeMessages finish ",getStatus().c_str()));
		
		// Close the rest of pipes all pipes
        if( isolated_process_as_tread )
        {
            sleep(1);  
            LM_T( LmtIsolated , ("Isolated process %s: Closing unused side of the pipe (thread mode) pipeFdPair1[1]:%d, pipeFdPair2[0]:%d ",getStatus().c_str(), pipeFdPair1[1], pipeFdPair2[0]));
            close(pipeFdPair1[1]);
            close(pipeFdPair2[0]);
        }
		
		LM_T( LmtIsolated , ("Isolated process %s: Closing the rest of fds of the pipe pipeFdPair1[0]:%d, pipeFdPair2[1]:%d ",getStatus().c_str(), pipeFdPair1[0], pipeFdPair2[1]));
		close(pipeFdPair1[0]);
		close(pipeFdPair2[1]);
		
		LM_T( LmtIsolated , ("Isolated process %s: waiting child pid=%d to finish ",getStatus().c_str(), pid));
		
		// Kill and wait the process
        if( !isolated_process_as_tread )
        {
            
            // Loop until the background thread is killed
            sub_status = "Waiting";
            int stat_loc;
            pid_t p=0;
            do
            {
                
                // Wait for
                for ( int i = 0 ; i < 100 ; i++)
                {
                    p = waitpid(pid, &stat_loc , WNOHANG );
                    if( p == pid)
                        break;
                    usleep(10000);
                }

                
                // Send a kill message if still not died
                if( p!= pid )
                {
                    LM_T(LmtIsolated,("Killing background process (%d) manually", pid));
                    kill( pid , SIGKILL );
                    
                    // Give the background process some air to die in peace
                    usleep(10000);
                }
                else
                {
                    if ( WIFEXITED(stat_loc) )
                    {
                        int s = WEXITSTATUS( stat_loc );
                        LM_T(LmtIsolated,("Background process (pid=%d) ended with exit with code %d",pid, s));
                        
                    }
                    else if( WIFSIGNALED( stat_loc ) )
                    {
                        int s = WTERMSIG( stat_loc );
                        LM_T(LmtIsolated,("Background process (pid=%d) ended with signal with signal %d",pid, s));
                        LM_W(("Background process (pid=%d) ended with signal with signal %d",pid, s));
                    }
                    else
                    {
                        LM_T(LmtIsolated,("Background process (pid=%d) 'crashed' with unknown reason", pid));
                        LM_E(("Background process (pid=%d) 'crashed' with unknown reason", pid));
                    }
                }
                
                LM_W(("Waiting for background process with pid %d" , pid));
                
            } while (p != pid);
        }
		
        LM_T( LmtIsolated , ("Isolated process %s: Finish ******************************************************************************************* ",getStatus().c_str()));
        
	}
    
    bool ProcessItemIsolated::processProcessPlatformMessage( samson::network::MessageProcessPlatform * message )
    {
        
        // Eval if the process has been canceled by ProcessManager, if so, it should return assap
        
        if( isProcessItemCanceled() )
        {
            error.set( "ProcessItem canceled");
	    LM_W(("ProcessItem canceled, operation:%d",  message->operation() ));
            
            // Send a kill message and finish
            samson::network::MessagePlatformProcess * response = new samson::network::MessagePlatformProcess();
            response->set_code( samson::network::MessagePlatformProcess_Code_code_kill );
            if (au::writeGPB(pipeFdPair2[1], response) != au::OK)
	    {
		    LM_E(("Error sending cancel response, code(%d),  (pipeFdPair2[1]:%d) ",  response->code(), pipeFdPair2[1]));
	    }
            delete response;
            
            // Finish since it has been canceled
            return true;
        }
        <commit_msg>SAMSON-291 More elegant way to wait to background processes<commit_after>#include <pthread.h>
#include <iostream>                 // std::cerr
#include <signal.h>                 // kill(.)
#include <stdlib.h>                 // exit()
#include <sys/types.h>      
#include <sys/wait.h>               // waitpid()

#include "logMsg/logMsg.h"                 // LM_*
#include "logMsg/traceLevels.h"            // LmtIsolated, etc. 

#include "au/gpb.h"                     // au::readGPB & au::writeGPB
#include "au/ErrorManager.h"
#include "au/Token.h"
#include "au/TokenTaker.h"

#include "engine/MemoryManager.h"			// engine::MemoryManager
#include "engine/Notification.h"            // engine::Notification

#include "samson/common/traces.h"                       // logFd
#include "samson/common/SamsonSetup.h"                // Goyo: ss:SamsonSetup
#include "samson/isolated/ProcessIsolated.h"         //Goyo: For defines WORKER_TASK_ITEM_CODE_FLUSH_BUFFER
#include "samson/common/NotificationMessages.h"     // notification_samson_worker_send_trace
#include "samson/common/MemoryTags.h"                 // MemoryInput , MemoryOutputNetwork, ...

#include "samson/network/iomMsgAwait.h"            // iomMsgAwait
#include "samson/network/Packet.h"                     // samson::Packet


#include "samson/isolated/ProcessItemIsolated.h"    // Own interface
#include "samson/isolated/SharedMemoryManager.h"    

#include "SharedMemoryItem.h"           // engine::SharedMemoryItem


namespace samson
{
    
    class PidElement
    {
        pid_t pid;
        au::Cronometer cronometer;
        int kill_count;
        
    public:
        
        PidElement( pid_t _pid )
        {
            pid = _pid;
            kill_count = 0;
        }
        
        bool wait()
        {
            int stat_loc;
            pid_t p = waitpid(pid, &stat_loc , WNOHANG );
            if( p == pid)
            {
                if ( WIFEXITED(stat_loc) )
                {
                    int s = WEXITSTATUS( stat_loc );
                    LM_T(LmtIsolated,("Background process (pid=%d) ended with exit with code %d",pid, s));
                    
                }
                else if( WIFSIGNALED( stat_loc ) )
                {
                    int s = WTERMSIG( stat_loc );
                    LM_T(LmtIsolated,("Background process (pid=%d) ended with signal with signal %d",pid, s));
                    LM_W(("Background process (pid=%d) ended with signal with signal %d",pid, s));
                }
                else
                {
                    LM_T(LmtIsolated,("Background process (pid=%d) 'crashed' with unknown reason", pid));
                    LM_E(("Background process (pid=%d) 'crashed' with unknown reason", pid));
                }
                
                
                return true;
            }

            if( cronometer.diffTime() > 3.0 ) // Three seconds to die
            {
                // Send a kill signal to this process
                LM_T(LmtIsolated,("Killing background process (%d) manually", pid));
                kill( pid , SIGKILL );
                
                kill_count++;
                cronometer.reset();
                
                if( kill_count > 3 )
                    LM_W(("Background process (pid %d) is not dying after %d kills" , pid , kill_count ));
                
            }
            
            
            
            return false;
        }
        
    };

    class PidCollection
    {
        au::Token token;
        std::set<PidElement*> pids;
        
        public:
        
        PidCollection() : token("PidCollection")
        {
            
        }
        
        void add( pid_t pid )
        {
            au::TokenTaker tt(&token);
            pids.insert( new PidElement(pid) );
            
            if( pids.size() > 20 )
                LM_W(("Waiting a high number of background process %d" , (int)  pids.size() ));
                                  
        }
        
        void clear()
        {
            au::TokenTaker tt(&token);
            
            std::set<PidElement*>::iterator it_pids;

            for ( it_pids = pids.begin() ; it_pids != pids.end() ;  )
            {
                PidElement* pid_element = (*it_pids);
                
                if( pid_element->wait() )
                {
                    delete pid_element;
                    pids.erase( it_pids++ );
                }
                else
                    ++it_pids;
                
            }
        }
        
        
    };
    
    
    PidCollection myPidCollection;
    
    
    // Static variable to active the "thread" mode of the background process execution
    bool ProcessItemIsolated::isolated_process_as_tread = false;
    	
	
	// Function to run by the backgroudn thread
	void* run_ProcessItemIsolated( void* p )
	{
        // Free resources automatically when this thread finish
        pthread_detach(pthread_self());
        
		ProcessItemIsolated* tmp = (ProcessItemIsolated*)p;
		tmp->runBackgroundProcessRun();
		return NULL;
	}
	
	ProcessItemIsolated::ProcessItemIsolated( std::string _description ) : ProcessItem( PI_PRIORITY_NORMAL_OPERATION ) 	
	{
        processItemIsolated_description = _description;
	}	

	ProcessItemIsolated::~ProcessItemIsolated()
    {
    }
	
    
	void ProcessItemIsolated::run()
	{
        LM_T( LmtIsolated , ("Isolated process %s: start ******************************************* ",getStatus().c_str()));
        
        
        if( isolated_process_as_tread )
            LM_T( LmtIsolated , ("Isolated process %s start in thread mode",getStatus().c_str()));
        else
            LM_T( LmtIsolated , ("Isolated process %s start in fork mode",getStatus().c_str()));
        			
		// Create a couple of pipes to communicate both process
		if ( pipe(pipeFdPair1) != 0 )
		{
			LM_E(("System error: not possible to create pipes when running this process"));
			error.set("System error: not possible to create pipes when running this process");
			return;
		}
		if ( pipe(pipeFdPair2) != 0 )
		{
			LM_E(("System error: not possible to create pipes when running this process"));
			error.set("System error: not possible to create pipes when running this process");
			LM_T( LmtIsolated, ("Isolated process %s: pipes closed: pipeFdPair1[0]:%d, pipeFdPair1[1]:%d\n", pipeFdPair1[0], pipeFdPair1[1]));
			close( pipeFdPair1[0] );
			close( pipeFdPair1[1] );
			return;
		}
		
		LM_T(LmtIsolated, ("Isolated process %s: pipes created. pipeFdPair1[0]:%d, pipeFdPair1[1]:%d, pipeFdPair2[0]:%d, pipeFdPair2[1]:%d\n",getStatus().c_str(), pipeFdPair1[0], pipeFdPair1[1], pipeFdPair2[0], pipeFdPair2[1]));
		LM_T( LmtIsolated , ("Isolated process %s: pipes created ",getStatus().c_str()));
		
		// Create the other process to run the other side
        pid_t pid = 0;
        if( isolated_process_as_tread )
        {
            ProcessItemIsolated *tmp = (this);
            pthread_t t;
            pthread_create(&t, NULL, run_ProcessItemIsolated, tmp);
        }
        else
        {
        	LM_T( LmtIsolated , ("Isolated process %s: father about to fork",getStatus().c_str()));
            pid = fork();
            if ( pid < 0 )
                LM_X(1,("Fork return an error"));
            
            if( pid == 0 )	// Children running the background process
            {
                runBackgroundProcessRun();
                LM_T(LmtIsolated,("Child in Background process finished, calling _exit that will close its side pipes: pipeFdPair1[1]:%d, pipeFdPair2[0]:%d\n", pipeFdPair1[1], pipeFdPair2[0]));
                _exit(1000);
            }
        }
		
        // Exchange all the necessary messages between background and foreground process
        LM_T( LmtIsolated , ("Isolated process %s: father runExchangeMessages start, child pid=%d ",getStatus().c_str(), pid));
        
        runExchangeMessages();
        
        LM_T( LmtIsolated , ("Isolated process %s: father runExchangeMessages finish ",getStatus().c_str()));
		
		// Close the rest of pipes all pipes
        if( isolated_process_as_tread )
        {
            sleep(1);  
            LM_T( LmtIsolated , ("Isolated process %s: Closing unused side of the pipe (thread mode) pipeFdPair1[1]:%d, pipeFdPair2[0]:%d ",getStatus().c_str(), pipeFdPair1[1], pipeFdPair2[0]));
            close(pipeFdPair1[1]);
            close(pipeFdPair2[0]);
        }
		
		LM_T( LmtIsolated , ("Isolated process %s: Closing the rest of fds of the pipe pipeFdPair1[0]:%d, pipeFdPair2[1]:%d ",getStatus().c_str(), pipeFdPair1[0], pipeFdPair2[1]));
		close(pipeFdPair1[0]);
		close(pipeFdPair2[1]);
		
		LM_T( LmtIsolated , ("Isolated process %s: waiting child pid=%d to finish ",getStatus().c_str(), pid));
		
		// Kill and wait the process
        if( !isolated_process_as_tread )
        {
            
            myPidCollection.add(pid);
            myPidCollection.clear();
        }
		
        LM_T( LmtIsolated , ("Isolated process %s: Finish ******************************************************************************************* ",getStatus().c_str()));
        
	}
    
    bool ProcessItemIsolated::processProcessPlatformMessage( samson::network::MessageProcessPlatform * message )
    {
        
        // Eval if the process has been canceled by ProcessManager, if so, it should return assap
        
        if( isProcessItemCanceled() )
        {
            error.set( "ProcessItem canceled");
	    LM_W(("ProcessItem canceled, operation:%d",  message->operation() ));
            
            // Send a kill message and finish
            samson::network::MessagePlatformProcess * response = new samson::network::MessagePlatformProcess();
            response->set_code( samson::network::MessagePlatformProcess_Code_code_kill );
            if (au::writeGPB(pipeFdPair2[1], response) != au::OK)
	    {
		    LM_E(("Error sending cancel response, code(%d),  (pipeFdPair2[1]:%d) ",  response->code(), pipeFdPair2[1]));
	    }
            delete response;
            
            // Finish since it has been canceled
            return true;
        }
        
{'fraction_non_alphanumeric': 0.12700829570513494, 'fraction_numerical': 0.009555812244040743, 'mean_word_length': 1.9226638023630505, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, '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': '6636856', 'n_tokens_mistral': 5947, 'n_tokens_neox': 5503, 'n_words': 1312}