/*
 * Copyright 2015 Dmitry Suplatov, Nina Popova, Sergey Zhumatiy, 
 *                Vladimir Voevodin, Vytas Švedas
 * 
 *  This program is free software: you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation, either version 3 of the License, or of
 *  any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 * Program: mpiWrapper
 * 
 * Version: v.2.1.2 2017-12-10
 * 
 * Author:  Dmitry Suplatov d.a.suplatov(at)belozersky.msu.ru
 *
 * Webpage: http://biokinet.belozersky.msu.ru/mpiwrapper
 * 
 * Ref:     D. Suplatov, N. Popova, S. Zhumatiy, V. Voevodin, V. Švedas (2016) Parallel 
 *          workflow manager for non-parallel bioinformatic applications to solve large-scale 
 *          biological problems on a supercomputer, J Bioinform Comput Biol, 14(2), 1641008
 *          doi:10.1142/S0219720016410080.
 * 
 * General: This software is intended to solve large computational problems which can be divided 
 *          into independent subtasks to be analyzed separately without interprocess communication
 *
 * Aim:     Provide an interface to execute single-cpu (i.e. non-parallel) implementations 
 *          of scientific algorithms in a parallel mode - on a Desktop station powered by 
 *          a multi-core CPU or a supercomputer
 * 
 * Idea:    Two specialized threads - one for MPI communication and one "worker" for task 
 *          execution - are invoked on each processing unit to avoid deadlock while using 
 *          blocking calls to MPI. The communicator thread on the root CPU is also an 
 *          administrator of the tasks queue - it reads tasks from the taskfile and assigns 
 *          them to free nodes in FIFO priority. Communicators on the non-root CPUs are in 
 *          charge of communicating with the administrator on the root CPU in order to 
 *          receive new tasks. Worker threads on each node execute the assigned tasks. 
 *          Process terminates when the queue gets empty and execution of already assigned 
 *          tasks has finished. 
 * 
 * Funding: This work was supported by the Russian Foundation for Basic Research 
 *          [grant #14-07-00437] and the Russian Science Foundation [grant #15-14-00069]
 */

#include <mpi.h>
#include <fstream>
#include <sstream>
#include <iostream>
#include <iomanip>
#include <string.h>
#include <vector>
#include <stdlib.h>
#include <algorithm>
#include <unistd.h>
#include <pthread.h>
#include <stdio.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <spawn.h>
#include <ctime>


using namespace std;

//Define the number of attempts to resubmit a task before giving up on it
#define MAX_TASK_FAILURE 10
//Define the number of failures MAX_NODE_FAILURE before the node gets blocked to NODE_BLOCK_TIME seconds
#define MAX_NODE_FAILURE 5
#define NODE_BLOCK_TIME 3600

//Use node0 as administrator
const int root = 0;
//Minimum sleep time - 0.01s 
int min_sleeptime = 10000;
//Maximum sleep time - 1s
int max_sleeptime = 1000000;
//Data structure to pass arguments to POSIX threads
typedef struct {char *file;} struct_t;
//Node status indicator (1 -> vacant, 0 -> busy)
int node_vacant;
//Buffer to store a task assigned to the current node
char *current_task = NULL;
//Status of the previous task (0 -> no previous task, 1 -> OK, 2 -> failed)
int prev_task_status;
//Task status indicator (1 -> command is written to buffer and ready to execute, 0 -> wait)
int task_status;
//Kill switch (1 -> terminate on the current node, 0 -> nothing to do)
int kill_switch;
//Statistics for node failure
std::vector<int> node_failed;
//Timestamp when the node was blocked because of too many failures
std::vector<int> node_block_time;
//Statistics for task failure (how many times each particular task has failed)
std::vector<int> task_failed;
//Save id of current task for every node
std::vector<int> node_taskid;
//Vector to mark task status (true -> in queue; false -> on nodes | completed | failed)
std::vector<bool> tasks_status;
//Total number of tasks
int stat_task_total = 0;
//Number of tasks completed successfully
int stat_task_success = 0;
//Number of resubmissions
int stat_task_resubmit = 0;
//Number of node failures
int stat_node_fail = 0;
//Number of node blocks
int stat_node_block = 0;

/**
 * Current date and time
 * @return String with date and time in YYYY-MM-DD HH:MM:SS format
 */
const std::string currentDateTime() {
    time_t     now = time(0);
    struct tm  tstruct;
    char       buf[80];
    tstruct = *localtime(&now);    
    strftime(buf, sizeof(buf), "%Y-%m-%d %X", &tstruct);
    return buf;
}

/**
 * Fork a command on a single node
 * @param task char array with shell command
 * @return status code (0 -> OK, !0 -> FAILED)
 */
int execute (char *task) {            
    int world_rank;
    MPI_Comm_rank(MPI_COMM_WORLD, &world_rank); 
    
    pid_t child_pid;
    int status;
    
    //Prepare command line arguments as a null-terminated array
    std::vector<char*> args;
    char* tasks = strtok(task, " ");
    while(tasks != NULL) {       
        args.push_back(tasks);
        tasks = strtok(NULL, " ");
    }
    args.push_back(NULL);
    
    if ((child_pid = vfork()) < 0) {        
        exit(1);
    }   
            
    //Child process
    if (child_pid == 0) {               
        //Execute program args[0] with arguments args[0], args[1], args[2], etc.
        execve(args[0], &args.front(), NULL);
        _exit(1);
    }   
    //Parent process
    else {
        //Wait for child process
        wait(&status);        
    }              
    
    //Return status message
    return status;
}

/**
 * Start communicator/administrator process
 * @param arg For root node -> filename to read tasks, for other nodes -> NULL
 * @return 
 */
void *admin(void *arg) {
    
    int world_rank;
    MPI_Comm_rank(MPI_COMM_WORLD, &world_rank); 

    int world_size;
    MPI_Comm_size(MPI_COMM_WORLD, &world_size);               
    
    //Administrative process on the root node has two features:
    //1. It reads tasks from a file and saves them to the queue. 
    //   This queue with tasks is, therefore, available on the root node only.
    //2. It administrates the tasks in the queue (i.e. assigned tasks to vacant 
    //   nodes and once the queue gets empty sends KILL signal to all nodes)
    
    if (world_rank == root) {
        
        //Retrieve file name 
        struct_t *ff = (struct_t *)arg;   
        char *name = ff->file;

        //Read tasks from file and fill the queue
        cout << "Info: Filling queue with tasks from file " << name << endl;    
        string data;            
        ifstream filename (name);            
        std::vector<string> queue;
        if (filename.is_open())
        {
            while (!filename.eof())
            {
                getline(filename, data);
                if (data!="") {queue.push_back(data);}
                else {cout << "Warning: Skipping empty line in the task file" <<endl;}
            }
        }
        else
        {
            cout << "Error: Unable to open file " << name << endl;
            MPI_Abort(MPI_COMM_WORLD,1);
        }
        filename.close();       
        cout << "Info: Queue contains " << queue.size() << " tasks" <<endl;
        if (world_size > queue.size()) {
            cout << "Warning: Number of nodes (" << world_size << ") is greater than number of tasks (" << queue.size() << ")" << endl;
        }
        
        stat_task_total = queue.size();
        
        //Wait for 1 second
        //usleep(1000000);
                          
        node_failed.reserve(world_size);
        node_block_time.reserve(world_size);
        node_taskid.reserve(world_size);
        
        //This vector is to store information whether each node has sent a message to root reporting the vacant status (and status of the previous task)
        std::vector<int> node_report;
        
        node_report.reserve(world_size);
        for (int i = 0; i < world_size; i++) {
            node_failed[i] = 0;
            node_block_time[i] = -1;
            node_taskid[i] = -1;
            node_report[i] = 0;
        }
        
        task_failed.reserve(queue.size());
        tasks_status.reserve(queue.size());
        for (int i = 0; i < queue.size(); i++) {
            task_failed[i] = 0;
            tasks_status[i] = true;
        }
        
        int tasks_completed = 0;
        int taskid = -1;
        int cur_sleeptime = min_sleeptime;
        bool queue_emtpy = false;

        //Queue to hold ids of blocked nodes in chronological order        
        std::vector<int> blocked_node_queue;
        
        //Assign tasks in the queue to available nodes until all tasks have been successfully executed
        while (tasks_completed != queue.size()) {
                      
            //Search for a vacant node
            int node_select=-1;                                
            
            //First check for root node            
            // - if it is vacant and not is not blocked given that the queue is not empty (first run condition)            
            if (node_vacant == 1 && node_taskid[root] != -1 && node_block_time[root] == -1) {
                node_report[root] = 1;
                node_select = root;                
            }
            // - if it is vacant and has been previously assigned a task which has not been reported yet given that root not is not blocked (regular run condition)
            else if (node_vacant == 1 && !queue_emtpy && node_block_time[root] == -1) {                
                node_select = root;                
            }
                        
            //Second check for incomming status messages from non-root nodes
            if (node_select == -1) {                                            
                int flag = 0;
                MPI_Status status;
                MPI_Iprobe(MPI_ANY_SOURCE, 9, MPI_COMM_WORLD, &flag, &status);                                            
                if (flag) {                    
                    int vacant;
                    int source = status.MPI_SOURCE;
                    MPI_Recv(&vacant, 1, MPI_INT, source, 9, MPI_COMM_WORLD, MPI_STATUS_IGNORE);                    
                    if (vacant == 1) {
                        node_select = source;       
                        node_report[node_select] = 1;                        
                    }
                    else {
                        cout << "Error: Internal error (message with tag 9 from node " << node_select << " has sent a non-vacant status message \"" << vacant << "\")" << endl;
                        MPI_Abort(MPI_COMM_WORLD,1);
                    }                
                }
            }
                        
            //Third check blocked nodes if the time has come to unblock them
            if (node_select == -1 && !blocked_node_queue.empty()) {            
                //Check the block time of the first node in queue
                int first_blocked_node = blocked_node_queue.front();
                
                //Check if the selected node is in the blocked state
                if (node_block_time[first_blocked_node] == -1) {
                    cout << "Error: Internal error (node " << first_blocked_node << " is marked as blocked in the blocked_node_queue but not in the node_block_time)" << endl;
                    MPI_Abort(MPI_COMM_WORLD,1);
                }
                
                //Check nodes in the blocked state which have already sent status messages
                time_t now = time(0);
                int cur_seconds = now;
                //time is up and we can use this node right now
                if (cur_seconds - node_block_time[first_blocked_node] >= NODE_BLOCK_TIME) {
                    cout << "Info: Node#" << first_blocked_node << " has been unblocked at " << currentDateTime() << endl;
                    node_block_time[first_blocked_node] = -1;
                    node_select=first_blocked_node;
                    //Errase the first node from the queue
                    blocked_node_queue.erase( blocked_node_queue.begin() );
                }
                //else - if the first node is yet blocked than all the other nodes are too
            }
                        
            //Fourth check if there are any nodes with no task assignment
            if (node_select == -1) {
                for (int i = 0; i < world_size; i++) {                
                    if (node_taskid[i] == -1 && node_block_time[i] == -1 && node_report[i] == 1) {
                        node_select = i;
                        break;
                    }
                }
            }
                                                              
            //If we have checked all nodes available and did not find a vacant node then we wait and check again
            if (node_select == -1) {
                usleep(cur_sleeptime); 
                cur_sleeptime = min(cur_sleeptime * 2, max_sleeptime);
                continue;
            }
            //else {cur_sleeptime = min_sleeptime;} the sleep time will be reset if both the task has been selected and the queue is not empty
                        
            //Learn what happened with the previous task on this node (unless checked previously)
            bool cur_node_report = false;            
            if (node_taskid[node_select] != -1) {
                cur_node_report = true;
                
                int previous_taskid = node_taskid[node_select];
                node_taskid[node_select] = -1;
                
                //Retrieve prev_status from the selected node
                int temp_prev_status = 0;
                if (node_select == root) {
                   temp_prev_status = prev_task_status;
                }
                else {
                    int request = 2;
                    MPI_Send(&request, 1, MPI_INT, node_select, 0, MPI_COMM_WORLD);
                    MPI_Recv(&temp_prev_status, 1, MPI_INT, node_select, 0, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
                }
                //Check prev_status
                //Previous task has completed successfully 
                if (temp_prev_status == 1) {                    
                    tasks_completed++;
                    stat_task_success++;
                    cout << "Info: Task#" << previous_taskid << " successfully completed on Node#" << node_select << " at " << currentDateTime()
                         << " (" << (queue.size() - tasks_completed) << " tasks left)" << endl;
                }
                //Previous task has failed
                else if (temp_prev_status == 2) {                
                    node_failed[node_select]++;
                    task_failed[previous_taskid]++;
                    stat_node_fail++;
                    
                    if (task_failed[previous_taskid] == MAX_TASK_FAILURE) {
                        cout << "Warning: Task#" << previous_taskid << " has failed on Node#" << node_select << " at " << currentDateTime()
                        << " (" << task_failed[previous_taskid] << " times in total) and will be removed from the queue" << endl;
                        tasks_completed++;
                    }                    
                    else {
                        cout << "Warning: Task#" << previous_taskid << " has failed on Node#" << node_select << " at " << currentDateTime()
                        << " (" << task_failed[previous_taskid] << " times in total) and will be resubmitted" << endl;
                        tasks_status[previous_taskid] = true;
                        stat_task_resubmit++;
                    }
                    
                    if (node_failed[node_select] == MAX_NODE_FAILURE) {
                        cout << "Warning: Node#" << node_select << " has failed " << node_failed[node_select]
                            << " times and will be blocked for " << NODE_BLOCK_TIME << " seconds starting from " << currentDateTime() << endl;
                        time_t now = time(0);
                        int cur_seconds = now;
                        node_block_time[node_select] = cur_seconds;
                        node_failed[node_select] = 0;
                        blocked_node_queue.push_back(node_select);
                        stat_node_block++;
                        continue;
                    }
                }
            }
                               
            //Do we have any more tasks left in the queue?
            int cur_taskid = -1;
            int cur_i = taskid+1;
            while (cur_taskid == -1) {
                if (cur_i == queue.size()) {cur_i = 0; continue;}                
                if (tasks_status[cur_i] == true) {cur_taskid = cur_i; break;}                
                if (cur_i == taskid) {break;}
                cur_i++;
            }
                        
            //No tasks left in queue (but some can still be on nodes and may return as failed)
            if (cur_taskid == -1) {
                queue_emtpy = true;
                
                if (!cur_node_report) {
                    usleep(cur_sleeptime); 
                    cur_sleeptime = min(cur_sleeptime * 2, max_sleeptime);
                }
                else {
                    cur_sleeptime = min_sleeptime;
                }
                continue;
            }   
            else {                
                queue_emtpy = false;
                taskid = cur_taskid;
            }
            
            //Here we reset the sleep time to minimum value - we have located both a vacant node and a task
            cur_sleeptime = min_sleeptime;
                      
            //Assign the selected task to the selected node
            string strtask = queue[taskid];	    
            int length = strtask.length();	                            
            if (node_select == root) {
                node_vacant = 0;
                current_task = new char[length+1];
                strcpy(current_task, strtask.c_str());
                task_status = 1;   
            }
            else {
                int request = 3;
                MPI_Send(&request, 1, MPI_INT, node_select, 0, MPI_COMM_WORLD);
                char *task = new char[length+1]; 
                strcpy(task, strtask.c_str());
                MPI_Send(task, length+1, MPI_CHAR, node_select, 1, MPI_COMM_WORLD);
                delete[] task;
            }
            cout << "Info: Task#" << taskid << " sent to Node#" << node_select << ": <BEGIN>" << strtask << "<END> at " << currentDateTime() << endl;
            node_taskid[node_select] = taskid;
            tasks_status[taskid] = false;
            node_report[node_select] = 0;
        }
        
        //When the queue is empty send KILL switch to all nodes
        cout << "Info: Queue is empty" << endl;
        //Statistics for time usage on nodes
        int request = 0;
        for (int i = 0; i < world_size; i++) {
            if (node_report[i] == 0) {
                cout << "Error: Internal error (node " << i << " has not reported its status but is receiving the kill switch)" << endl;
                MPI_Abort(MPI_COMM_WORLD,1);
            }
            cout << "Info: Sending KILL switch to Node#" << i << endl;
            //If node#i is root - set kill switch in shared memory
            if (i == root) {kill_switch = 1; continue;} 
            //If node#i is not root - use MPI communicator
            MPI_Send(&request, 1, MPI_INT, i, 0, MPI_COMM_WORLD); 
        }
        //Terminate administrative thread on the root node
        pthread_exit(NULL); 
    }   
    
    //Communicator on a non-root node communicates with the administrator
    //on the root node in order to receive new tasks
    else {
                
        int cur_sleeptime = min_sleeptime;
        while (true) {            
            if (node_vacant != 1) {
                usleep(cur_sleeptime); 
                cur_sleeptime = min(cur_sleeptime * 2, max_sleeptime);
                continue;
            }
            cur_sleeptime = min_sleeptime;
            
            //If we are here it means that this node is vacant
            
            //Send status
            MPI_Send(&node_vacant, 1, MPI_INT, root, 9, MPI_COMM_WORLD);
            
            //Start the receive mode -- wait for requests from the administrator
            while (true) {
                //Receive request from the root-node administrator
                int request;
                MPI_Recv(&request, 1, MPI_INT, root, 0, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
                
                //Terminate if kill_switch received 
                if (request == 0) {
                    kill_switch = 1;
                    pthread_exit(NULL);
                }
                //Send node occupancy status -- OBSOLETE
                else if (request == 1) {    
                    MPI_Send(&node_vacant, 1, MPI_INT, root, 0, MPI_COMM_WORLD);
                    continue;
                }
                //Send status for the previous task 
                else if (request == 2) {
                    MPI_Send(&prev_task_status, 1, MPI_INT, root, 0, MPI_COMM_WORLD);
                    continue;
                }
                //Recieve task
                else if (request == 3) {
                    node_vacant = 0; //Change node occupancy status to "occupied"
                    int char_amount;
                    MPI_Status status;        
                    MPI_Probe(root, 1, MPI_COMM_WORLD, &status);
                    MPI_Get_count(&status, MPI_CHAR, &char_amount);                        
                    current_task = new char[char_amount];       
                    MPI_Recv(current_task, char_amount, MPI_CHAR, root, 1, MPI_COMM_WORLD, MPI_STATUS_IGNORE);                           
                    //cout << "Info: Node#" << world_rank << " received task: <BEGIN>" << current_task << "<END>" << endl;
                    task_status = 1; //Change task status to "ready to execute"
                    break;
                }
            }
        }
    }        
}

int main(int argc, char * argv[])
{ 
    int provided;
    MPI_Init_thread(&argc, &argv, MPI_THREAD_MULTIPLE, &provided);
  
    int world_size;
    MPI_Comm_size(MPI_COMM_WORLD, &world_size);        
    
    int world_rank;
    MPI_Comm_rank(MPI_COMM_WORLD, &world_rank);    
    
    if (world_rank == root) {
        
        //Print header
        cout << "" << endl;
        cout << "        :-) :-) :-) MPILAUNCHER (-: (-: (-: " << endl;
        cout << "             Parallel workflow manager      " << endl;
        cout << "           for non-parallel applications    " << endl;
        cout << "                v.2.1.2 2017-12-10          " << endl;
        cout << "" << endl;
        cout << "  If you find this software or its results usefull please cite" << endl;
        cout << "  Suplatov D, Popova N, Zhumatiy S, Voevodin V, Švedas V. (2016)" << endl;
        cout << "  Parallel workflow manager for non-parallel bioinformatic " << endl;
	cout << "  applications to solve large-scale biological problems on a " << endl;
	cout << "  supercomputer. J. Bioinform. Comput. Biol., 14(02), 1641008." << endl;
        cout << "" << endl;
        cout << "  Acknowledgements: This work was supported by the Russian Foundation" << endl;
	cout << "  for Basic Research [grant #14-07-00437] and the Russian Science " << endl;
	cout << "  Foundation [grant #15-14-00069]" << endl;
        cout << "" << endl;
        
        //Print help
        if (!argv[1]) 
        {   
	    cout << "  General usage:" << endl;
            cout << "    mpilauncher [options] /path/to/mpiWrapper /path/to/taskfile.txt" << endl;
	    cout << "    The \"mpilauncher\" depends on you particular system and its [options] "<< endl;
	    cout << "    have to specify the number of CPUs (CPU cores) to be used by the "<<endl;
	    cout << "    mpiWrapper to process all tasks given in the taskfile.txt" << endl;
	    cout << "" << endl;
	    cout << "  Basic examples:" << endl;
	    cout << "    Launch mpiWrapper locally on 8 CPU cores" << endl;
	    cout << "      mpirun -hosts localhost -np 8 " << argv[0] << " taskfile.txt" << endl;	
	    cout << "" << endl;
	    cout << "    Launch mpiWrapper compiled with IntelMPI on 64 CPU cores of a "<<endl;
	    cout << "    supercomputer with sbatch and impi batch script" << endl;
	    cout << "      sbatch -np 64 impi " << argv[0] << " taskfile.txt" << endl;
	    cout << "" << endl;
	    cout << "    Launch mpiWrapper on 64 CPU cores of a supercomputer using cleo " <<endl;
	    cout << "    queue manager" << endl;
	    cout << "      cleo-submit -np 64 " << argv[0] << " taskfile.txt" << endl;
            cout << "" << endl;
            return EXIT_SUCCESS;
        }
    }
    
    //Undocumented command-line arguments
    if (argc > 2) {
        istringstream ss(argv[2]);
        int x;
        if (!(ss >> x)) {cout << "Error: Invalid number for min_sleeptime " << argv[2] << '\n'; exit (EXIT_FAILURE);}
        min_sleeptime = x;
        if (world_rank == root) {cout << "Info: Updating min_sleeptime to " << min_sleeptime << " microseconds" << endl;}
    }
    if (argc > 3) {
        istringstream ss(argv[3]);
        int x;
        if (!(ss >> x)) {cout << "Error: Invalid number for max_sleeptime " << argv[3] << '\n'; exit (EXIT_FAILURE);}
        max_sleeptime = x;
        if (world_rank == root) {cout << "Info: Updating max_sleeptime to " << max_sleeptime << " microseconds" << endl;}
    }        
    
    //Wait for 1 second
    //usleep(1000000);
    
    //Initialize node and task status on all nodes
    node_vacant = 1;
    task_status = 0;
    prev_task_status = 0;
    kill_switch = 0;
        
    //Malloc one pthread
    pthread_t *threads = new pthread_t[1];
    
    //Start administrative thread on root node (provide path to task file)
    if (world_rank == root) {
        cout << "Info: Started at " << currentDateTime() << endl;
        cout << "Info: MPI World contains " << world_size << " CPUs" << endl;
        char *name = argv[1];
        struct_t *fn = new struct_t[1];
        fn[0].file = name;
        pthread_create(&threads[0], NULL, admin, (void *)&fn[0]);
    }
    //Start administrative thread on non-root node
    else {
        pthread_create(&threads[0], NULL, admin, NULL);
    }           
    
    //Worker (starts on each node) - loop until kill switch received by the administrative thread
    int cur_sleeptime = min_sleeptime;
    while (true) {
        //If a new task has arrived - execute
        if (task_status == 1) {
            if (execute(current_task) == 0) {
                //Successfully completed
                prev_task_status = 1;
            }
            else {
                //Failed (_method_ has returned a non-zero exit status)
                prev_task_status = 2;                
            }
            
            delete[] current_task;
            //current_task=NULL;
            task_status = 0;
            node_vacant = 1;
            cur_sleeptime = min_sleeptime;
        }
        //If no new tasks - sleep
        else if (task_status == 0) {
            if (kill_switch == 1) {
                //if (current_task != NULL) {  
		//    cout << "Error: Bug#1 Kill switch received but the task buffer is not empty on node#" << world_rank << endl;
		//    MPI_Abort(MPI_COMM_WORLD,1);
                //}
                break;
            }
            usleep(cur_sleeptime); 
            cur_sleeptime = min(cur_sleeptime * 2, max_sleeptime); 
        }        
    }
    
    //Wait for administrative thread to finish (if it has`t done so yet)
    pthread_join(threads[0], NULL);
    delete[] threads;
    
    cout << "Info: Node#" << world_rank << " has terminated normally" << endl;        
    
    //End MPI
    MPI_Finalize();            
    
    if (world_rank == root) {
        cout << "" << endl;
        cout << "Statistics" << endl;
        cout << "-----------------------------" << endl;
        cout << "Total tasks processed: " << stat_task_total << endl;
        cout << "Tasks processed successfully: " << stat_task_success << endl;
        cout << "Resubmitted on failure count: " << stat_task_resubmit << endl;
        cout << "Node failure count: " << stat_node_fail << endl;
        cout << "Node block count: " << stat_node_block << endl;
        cout << "-----------------------------" << endl;
        cout << "" << endl;
	cout << "If you find this program or its results useful please cite" << endl;
	cout << "Suplatov et al. (2016) J Bioinform Comput Biol, 14(2), 1641008" << endl;
        cout << "" << endl;
	cout << "Info: Ended at " << currentDateTime() << endl;
    }
}