Cosmos::Support::CommandQueue Class Reference

Class to manage information about a queue of Events. More...

#include <command_queue.h>

Collaboration diagram for Cosmos::Support::CommandQueue:

Public Member Functions
	~CommandQueue ()
	Ensure all threads are joined before destruction. More...
size_t	join_event_threads ()
	Join all threads spawn and empty our vector. More...
size_t	get_event_size ()
	Retrieve the size of the queue. More...
Event &	get_event (int i)
	Retrieve an Event by its position in the queue. More...
size_t	get_command_size ()
	Retrieve the size of the queue. More...
Event &	get_command (int i)
	Retrieve an Event by its position in the queue. More...
int32_t	load_commands (string incoming_dir)
	Load queue of Events from a file. More...
int32_t	save_commands (string temp_dir, string name=".queue")
	Save the queue of Events to a file. More...
int32_t	restore_commands (string temp_dir, string name=".queue")
	Restore the queue of Events from a file. More...
int32_t	run_command (Event &cmd, string nodename, double logdate_exec)
	Run the given Event. More...
int32_t	run_commands (Agent *agent, string nodename, double logdate_exec)
	Traverse the entire queue of Events, clearing those that have finished. More...
int32_t	add_command (Event &c)
	Remove all matching Event from the queue. More...
int32_t	del_command (Event &c)
	Remove all matching Event from the queue. More...
int32_t	del_command (int pos)
	Remove Event from the queue based on position. More...
void	sort ()
	Sort the Events in the queue by Event exectution time. More...

Private Attributes
std::list< Event >	commands
std::deque< Event >	events
vector< std::thread >	event_threads
bool	queue_changed = false
bool	queue_blocked = false

Friends
std::ostream &	operator<< (std::ostream &out, CommandQueue &cmdq)
	Extraction operator. More...

Detailed Description

Class to manage information about a queue of Events.

Constructor & Destructor Documentation

Cosmos::Support::CommandQueue::~CommandQueue

(

)

Ensure all threads are joined before destruction.

{ join_event_threads(); }

Member Function Documentation

size_t Cosmos::Support::CommandQueue::join_event_threads

(

)

Join all threads spawn and empty our vector.

        {
//            static auto join_event = [] (std::thread &t) {
//                t.join();
//            };

//            std::for_each(event_threads.begin(), event_threads.end(), join_event);
//            event_threads.clear();
            size_t count = 0;
            for(uint16_t i=0; i<event_threads.size(); ++i)
            {
                if (event_threads[i].joinable())
                {
                    event_threads[i].join();
                    ++count;
                }
            }

            if (count == event_threads.size())
            {
                event_threads.clear();
                return count;
            }
            else
            {
                return event_threads.size();
            }
        }

size_t Cosmos::Support::CommandQueue::get_event_size ( )

inline

Retrieve the size of the queue.

Returns

The size of the queue

{ return events.size(); }

Event& Cosmos::Support::CommandQueue::get_event ( int  i )

inline

Retrieve an Event by its position in the queue.

Parameters

i	Integer representing the position in the queue

Returns

Reference to the ith Event

            {
                std::deque<Event>::iterator ii = events.begin();
                std::advance(ii,i);
                return *ii;
            }

size_t Cosmos::Support::CommandQueue::get_command_size ( )

inline

Retrieve the size of the queue.

Returns

The size of the queue

{ return commands.size(); }

Event& Cosmos::Support::CommandQueue::get_command ( int  i )

inline

Retrieve an Event by its position in the queue.

Parameters

i	Integer representing the position in the queue

Returns

Reference to the ith Event

            {
                std::list<Event>::iterator ii = commands.begin();
                std::advance(ii,i);
                return *ii;
            }

int32_t Cosmos::Support::CommandQueue::load_commands

(

string

incoming_dir

)

Load queue of Events from a file.

Loads new commands from *.command files located in the incoming directory.

Reads new Events from *.command files in the incoming directory, adds them to the queue of Events, and deletes the *.command files. Events in the queue are then sorted by their execution time.

Parameters

incoming_dir

Directory where the .queue file will be read from

Commands are loaded into the global CommandQueue object (cmd_queue), .command files are removed, and the command list is sorted by utc.

Parameters

incoming_dir

Directory where the .command files will be read from

        {
            DIR *dir = nullptr;
            struct dirent *dir_entry = nullptr;

            // open the incoming directory
            if ((dir = opendir((char *)incoming_dir.c_str())) == nullptr)
            {
                return GENERAL_ERROR_OPEN;
            }

            // cycle through all the file names in the incoming directory
            while((dir_entry = readdir(dir)) != nullptr)
            {
                string filename = dir_entry->d_name;

                if (filename.find(".command") != string::npos)
                {

                    string infilepath = incoming_dir + filename;
                    std::ifstream infile(infilepath.c_str());
                    if(!infile.is_open())
                    {
//                        std::cout<<"unable to read file <"<<infilepath<<">"<<std::endl;
                        continue;
                    }

                    //file is open for reading commands
                    string line;
                    Event cmd;

                    while(getline(infile,line))
                    {
                        cmd.set_command(line);

                        if(cmd.is_command())
                        {
//                            std::cout << "Command added: " << cmd;
                            add_command(cmd);
                        }
//                        else
//                            std::cout<<"Not a command!"<<std::endl;
                    }
                    infile.close();

                    //remove the .command file from incoming directory
                    if(remove(infilepath.c_str()))  {
//                        std::cout<<"unable to delete file <"<<filename<<">"<<std::endl;
                        continue;
                    }

//                    std::cout<<"\nThe size of the command queue is: "<< get_command_size()<<std::endl;
                }
            }

            sort();

            closedir(dir);

            return 0;
        }

int32_t Cosmos::Support::CommandQueue::save_commands	(	string	temp_dir,
		string	name = ".queue"
	)

Save the queue of Events to a file.

Save the queue of Events to the file temp_dir/.queue

\param  temp_dir    Directory where the .queue file will be written

Commands are taken from the global command queue Command queue is sorted by utc after loading

Parameters

temp_dir	Directory where the .queue file will be written
name	File where the .queue will be written

        {
            if (!queue_changed)
            {
                return 0;
            }
            queue_changed = false;

            // Save previous queue
            rename((temp_dir+name).c_str(), (temp_dir+name+'.'+(utc2unixdate(currentmjd()))).c_str());

            // Open the outgoing file
            FILE *fd = fopen((temp_dir+name).c_str(), "w");
            if (fd != nullptr)
            {
                for (Event cmd: commands)
                {
                    fprintf(fd, "%s\n", cmd.get_event_string().c_str());
                }
                fclose(fd);
            }
            return static_cast<int32_t>(commands.size());
        }

int32_t Cosmos::Support::CommandQueue::restore_commands	(	string	temp_dir,
		string	name = ".queue"
	)

Restore the queue of Events from a file.

Save the queue of Events to the file temp_dir/.queue

\param  temp_dir    Directory where the .queue file will be written

Commands are taken from the global command queue Command queue is sorted by utc after loading

Parameters

temp_dir	Directory where the .queue file will be read from
name	File where the .queue will be read from

        {
            queue_changed = false;

            // Reload existing queue
            string infilepath = temp_dir + name;
            std::ifstream infile(infilepath.c_str());
            if(infile.is_open())
            {
                //file is open for reading commands
                string line;
                Event cmd;

                while(std::getline(infile,line))
                {
                    //cmd.set_command(line, agent);
                    cmd.set_command(line);

                    if(cmd.is_command())
                    {
                        add_command(cmd);
                    }
                }
                infile.close();
            }
            return static_cast<int32_t>(commands.size());
        }

int32_t Cosmos::Support::CommandQueue::run_command	(	Event &	cmd,
		string	node_name,
		double	logdate_exec
	)

Run the given Event.

Execute an event using ford(). For each event run, the time of execution (utcexec) is set, the flag EVENT_FLAG_ACTUAL is set to true, and this updated command information is logged to the OUTPUT directory.

\param  cmd Reference to event to run
\param  nodename    Name of node
\param  logdate_exec    Time of execution (for logging purposes)

Executes a command in a separate shell (system) using threads. For each command run, the time of execution (utcexec) is set, the flag EVENT_FLAG_ACTUAL is set to true, and this updated command information is logged to the OUTPUT directory.

Parameters

cmd	Reference to event to run
nodename	Name of node
logdate_exec	Time of execution (for logging purposes)

        {
            queue_changed = true;

            // set time executed & actual flag
            cmd.set_utcexec();
            cmd.set_actual();

            string outpath = data_type_path(node_name, "temp", "exec", logdate_exec, "out");
            char command_line[100];
            strcpy(command_line, cmd.get_data().c_str());

            // We keep track of all threads spawned to join before moving log files.
            event_threads.push_back(std::thread([=] () {
                int devn, prev_stdin, prev_stdout, prev_stderr;
                if (outpath.empty()) {
                    devn = open("/dev/null", O_RDWR);
                }
                else {
                    devn = open(outpath.c_str(), O_CREAT|O_WRONLY|O_APPEND, 00666);
                }

                prev_stdin = dup(STDIN_FILENO);
                prev_stdout = dup(STDOUT_FILENO);
                prev_stderr = dup(STDERR_FILENO);

                // Redirect all output our executed command.
                dup2(devn, STDIN_FILENO);
                dup2(devn, STDOUT_FILENO);
                dup2(devn, STDERR_FILENO);
                close(devn);

                // Execute the command.
                system(command_line);

                dup2(prev_stdin, STDIN_FILENO);
                dup2(prev_stdout, STDOUT_FILENO);
                dup2(prev_stderr, STDERR_FILENO);
                close(prev_stdin);
                close(prev_stdout);
                close(prev_stderr);
            }));

            //#if defined(COSMOS_WIN_OS)
            //  char command_line[100];
            //    strcpy(command_line, cmd.get_data().c_str());
            //
            //  STARTUPINFOA si;
            //  PROCESS_INFORMATION pi;
            //
            //  ZeroMemory( &si, sizeof(si) );
            //  si.cb = sizeof(si);
            //  ZeroMemory( &pi, sizeof(pi) );
            //
            //  if (CreateProcessA(NULL, (LPSTR) command_line, NULL, NULL, FALSE, 0, NULL, NULL, &si, &pi))
            //  {
            //      //      int32_t pid = pi.dwProcessId;
            //      CloseHandle( pi.hProcess );
            //      CloseHandle( pi.hThread );
            //  }
            //#else
            //    int32_t pid = fork(); // Fork paradigm copies ENTIRE process space, leads to errors when exiting child. Now using threads.
            //
            //    char *words[MAXCOMMANDWORD];
            //    string_parse((char *)cmd.get_data().c_str(), words, MAXCOMMANDWORD);
            //    string outpath = data_type_path(node_name, "temp", "exec", logdate_exec, "out");
            //    if (pid != 0) {
            //        signal(SIGCHLD, SIG_IGN); // Ensure no zombies.
            //    }
            //    else {
            //        int devn;
            //        if (outpath.empty()) {
            //            devn = open("/dev/null",O_RDWR);
            //        }
            //        else {
            //            devn = open(outpath.c_str(), O_CREAT|O_WRONLY|O_APPEND, 00666);
            //        }
            //      dup2(devn, STDIN_FILENO);
            //      dup2(devn, STDOUT_FILENO);
            //      dup2(devn, STDERR_FILENO);
            //      close(devn);
            //
            //        // Execute the command.
            //        execvp(words[0], &(words[1]));
            //        fflush(stdout);
            //        exit (0);
            //    }
            //
            //#endif

            // log to event file
            log_write(node_name, "exec", logdate_exec, "event", cmd.get_event_string().c_str());
            return 0;
        }

int32_t Cosmos::Support::CommandQueue::run_commands	(	Agent *	agent,
		string	node_name,
		double	logdate_exec
	)

Traverse the entire queue of Events, clearing those that have finished.

Traverse the entire queue of Events, and run those which qualify.

An Event qualifies to be cleared if its thread could be joined, or if it has been running for more than 1 minute.

\param  agent   Pointer to Agent object (for call to condition_true(..))
\param  nodename    Name of the node
\param  logdate_exec    Time of execution (for logging purposes)

Traverse the entire queue of Events, and run those which qualify.

An Event only qualifies to run if the current time is greater than or equal to the execution time of the Event. Further, if the Event is conditional, then the Event condition must be true.

\param  agent   Pointer to Agent object (for call to condition_true(..))
\param  nodename    Name of the node
\param  logdate_exec    Time of execution (for logging purposes)

An Event only qualifies to run if the current time is greater than or equal to the execution time of the Event. Further, if the Event is conditional, then the Event condition must be true.

\param  agent   Pointer to Agent object (for call to condition_true(..))
\param  nodename    Name of the node
\param  logdate_exec    Time of execution (for logging purposes)

        {
            for(std::list<Event>::iterator ii = commands.begin(); ii != commands.end(); ++ii)
            {
                // if command is not solo, or event_threads queue is empty
                if (event_threads.empty() || (!ii->is_solo() && !queue_blocked))
                {
                    if (ii->is_solo())
                    {
                        queue_blocked = true;
                    }
                    else
                    {
                        queue_blocked = false;
                    }
                    // if command is ready
                    if (ii->is_ready())
                    {
                        // if command is conditional
                        if (ii->is_conditional())
                        {
                            // if command condition is true
                            if(ii->condition_true(agent->cinfo))
                            {
                                // if command is repeatable
                                if(ii->is_repeat())
                                {
                                    // if command has not already run
//                                    if(!ii->already_ran)
                                    if (!ii->is_alreadyrun())
                                    {
                                        strncpy(agent->cinfo->node.lastevent, ii->name.c_str(), COSMOS_MAX_NAME);
                                        agent->cinfo->node.lasteventutc = currentmjd();
                                        run_command(*ii, node_name, logdate_exec);
                                        ii->set_alreadyrun(true);
                                        events.push_back(*ii);
                                        if (events.size() > 10)
                                        {
                                            events.pop_front();
                                        }
//                                        ii->already_ran = true;
                                        break;
                                    }
                                }
                                // else command is non-repeatable
                                else
                                {
                                    run_command(*ii, node_name, logdate_exec);
                                    events.push_back(*ii);
                                    if (events.size() > 10)
                                    {
                                        events.pop_front();
                                    }
                                    commands.erase(ii--);
                                    break;
                                }
                            }
                            // else command condition is false
                            else
                            {
                                ii->set_alreadyrun(false);
                            }
                        }
                        // else command is non-conditional
                        else
                        {
                            run_command(*ii, node_name, logdate_exec);
                            events.push_back(*ii);
                            if (events.size() > 10)
                            {
                                events.pop_front();
                            }
                            commands.erase(ii--);
                        }
                    }
                }
                else
                {
                    ;//cout<<"This command is *NOT* ready to run! ";
                }
            }
            return static_cast<int32_t>(commands.size());
        }

int32_t Cosmos::Support::CommandQueue::add_command

(

Event &

c

)

Remove all matching Event from the queue.

Parameters

c	Event to remove

Returns

The number of Events removed

JIMNOTE: this only adds given Event to the queue if the Event has flag for EVENT_TYPE_COMMAND set to true

        {
            // Replace if it matches an existing command, otherwise add to queue
            for (std::list<Event>::iterator ii = commands.begin(); ii != commands.end(); ++ii) {
                if (c.get_name() == ii->get_name()) {
                    *ii = c;
                    queue_changed = true;
                    return 0;
                }
            }

            commands.push_back(c);
            queue_changed = true;
            return 1;
        }

int32_t Cosmos::Support::CommandQueue::del_command

(

Event &

c

)

Remove all matching Event from the queue.

Parameters

c	Event to remove

Returns

The number of Events removed

This function only removes events from the queue if the are exactly equal to the given Event.

        {
            size_t prev_sz = commands.size();
            for (std::list<Event>::iterator ii = commands.begin();
                 ii != commands.end();
                 ++ii) {
                if (c == *ii) {
                    commands.erase(ii--);
                }
            }

            queue_changed = true;
            return static_cast<int32_t>(prev_sz - commands.size());

        }

int32_t Cosmos::Support::CommandQueue::del_command

(

int

pos

)

Remove Event from the queue based on position.

Parameters

pos	Position of event to remove

Returns

The number of Events removed

This function removes events based on their queue position (0-indexed).

        {
            size_t prev_sz = commands.size();
            std::list<Event>::iterator b = commands.begin();

            std::advance(b, pos);
            commands.erase(b);
            queue_changed = true;
            return static_cast<int32_t>(prev_sz - commands.size());
        }

void Cosmos::Support::CommandQueue::sort ( )

inline

Sort the Events in the queue by Event exectution time.

This function is called after new Events are loaded.

{ commands.sort([](Event & c1, Event & c2) { return c1.getTime() < c2.getTime(); });    }

Friends And Related Function Documentation

std::ostream& operator<< ( std::ostream &  out, CommandQueue &  cmdq  )

friend

Extraction operator.

Parameters

out	Reference to ostream
cmdq	Reference to CommandQueue (JIMNOTE: should be const, ya?)

Returns

Reference to modified ostream

Writes the given CommandQueue to the given output stream (in JSON format) and returns a reference to the modified ostream.

        {
            for(std::list<Event>::iterator ii = cmdq.commands.begin(); ii != cmdq.commands.end(); ++ii)
                out << *ii << std::endl;
            return out;
        }

Member Data Documentation

std::list<Event> Cosmos::Support::CommandQueue::commands

private

An std::list of members of the Event class to be run

std::deque<Event> Cosmos::Support::CommandQueue::events

private

An std::queue of members of the Event class that have run

vector<std::thread> Cosmos::Support::CommandQueue::event_threads

private

A vector of all threads spawned to run events

bool Cosmos::Support::CommandQueue::queue_changed = false

private

A boolean indicator that the queue has changed

bool Cosmos::Support::CommandQueue::queue_blocked = false

private

---

The documentation for this class was generated from the following files:

• command_queue.h
• command_queue.cpp