<commit_before>/*
 *  Coordinator_Worker_Scheme.hpp
 */

//          Copyright <NAME> 2011.
// Distributed under the Boost Software License, Version 1.0.
//    (See accompanying file LICENSE_1_0.txt or copy at
//          http://www.boost.org/LICENSE_1_0.txt)

/************************************************
 *        Coordinator_Worker_Scheme             *
 ************************************************/

inline Coordinator_Worker_Scheme::Coordinator_Worker_Scheme ( int argc, char * argv [] ) 
: argc ( argc ), argv ( argv ) {}

/* Coordinator_Worker_Scheme::run() */
inline void Coordinator_Worker_Scheme::run ( Coordinator_Worker_Process * my_process,
                                             Communicator * my_communicator ) {
  // Set the command line arguments for the process
  my_process -> command_line ( argc, argv );
  // Determine if Coordinator or Worker
  if ( my_communicator -> coordinating () ) {
    run_coordinator ( my_process, my_communicator );
  } else {
    run_worker ( my_process, my_communicator );
  }
}

/* Coordinator_Worker_Scheme::run_coordinator() */
inline void Coordinator_Worker_Scheme::run_coordinator ( Coordinator_Worker_Process * my_process,
                                                         Communicator * my_communicator ) {
  std::cout << "COORD: Initialize.\n";
  my_process -> initialize ();
  int pending = 0;
  bool retiring = false;  
	while ( not retiring || pending > 0 ) { 
		/* Receive a message */
    std::cout << "COORD: Receive a message.\n";
    Message incoming;
    Channel worker;
		my_communicator -> receive ( &incoming, 
                                 &worker, 
                                 my_communicator -> ANYTAG, 
                                 my_communicator -> ANYSOURCE );
		/* Deal with READY. */
		if ( incoming . tag == READY ) {
      std::cout << "COORD: Deal with READY.\n";
      Message job_message;
      /* Prepare a job. (if we cannot, break )*/
      int prepare_status = my_process -> prepare ( job_message );
			/* Check the status of prepare and act accordingly. */
      if ( prepare_status == 0 )  {
				job_message . tag = JOB;
        ++ pending;
        my_communicator -> send ( job_message, worker );
			} else {
        retiring = true;
      }
		} /* if */
		
		/* Deal with a RESULTS message */
		if ( incoming . tag == RESULTS ) {
      std::cout << "COORD: Deal with RESULTS.\n";
      -- pending;
      my_process -> accept ( incoming ); 
    }
  }
  
  // broadcast retire message
  std::cout << "COORD: Broadcast RETIRE.\n";
  Message retire_message;
  my_process -> prepare ( retire_message );
  retire_message . tag = RETIRE;
  my_communicator -> broadcast ( retire_message );
  
  std::cout << "COORD: Finalize.\n";
  my_process -> finalize ();
}

/* Coordinator_Worker_Process::run_worker() */
inline void Coordinator_Worker_Scheme::run_worker ( Coordinator_Worker_Process * my_process,
                                                    Communicator * my_communicator ) {
  /* Main loop */
  Channel boss = my_communicator -> DIRECTOR;
  while ( true ) {
    /* Send a READY message. */
    //std::cout << "Worker_Base::run -- send a READY message.\n";
    Message ready;
    ready . tag = READY;
    my_communicator -> send ( ready, boss );
    /* Receive a job. */
    //std::cout << "Worker_Base::run -- receive a job\n";
    Message job_message;
		my_communicator -> receive ( &job_message, &boss, 
                                 my_communicator -> ANYTAG, my_communicator -> ANYSOURCE );
    /* Quit if it is a retire job. */
    if ( job_message . tag == RETIRE ) {
      //std::cout << "Worker_Base::run -- retiring\n";
      break;
    }
    /* Work the job. */
    //std::cout << "Worker_Base::run -- work a job\n";
    Message result_message;
    my_process -> work ( result_message, job_message );
		result_message . tag = RESULTS;
    /* Send the results back to the coordinator. */
    //std::cout << "Worker_Base::run -- report results\n";
    my_communicator -> send ( result_message, boss );
  }
}<commit_msg>Added USEIDLEWORKERS NUMBER_OF_WORKERS option for locking out part of cluster using scheduler (experimental feature)<commit_after>/*
 *  Coordinator_Worker_Scheme.hpp
 */

//          Copyright <NAME> 2011.
// Distributed under the Boost Software License, Version 1.0.
//    (See accompanying file LICENSE_1_0.txt or copy at
//          http://www.boost.org/LICENSE_1_0.txt)

/************************************************
 *        Coordinator_Worker_Scheme             *
 ************************************************/

inline Coordinator_Worker_Scheme::Coordinator_Worker_Scheme ( int argc, char * argv [] ) 
: argc ( argc ), argv ( argv ) {}

/* Coordinator_Worker_Scheme::run() */
inline void Coordinator_Worker_Scheme::run ( Coordinator_Worker_Process * my_process,
                                             Communicator * my_communicator ) {
  // Set the command line arguments for the process
  my_process -> command_line ( argc, argv );
  // Determine if Coordinator or Worker
  if ( my_communicator -> coordinating () ) {
    run_coordinator ( my_process, my_communicator );
  } else {
    run_worker ( my_process, my_communicator );
  }
}

/* Coordinator_Worker_Scheme::run_coordinator() */
inline void Coordinator_Worker_Scheme::run_coordinator ( Coordinator_Worker_Process * my_process,
                                                         Communicator * my_communicator ) {
  std::cout << "COORD: Initialize.\n";
  my_process -> initialize ();
  int pending = 0;
  bool retiring = false;  
	while ( not retiring || pending > 0 ) { 
		/* Receive a message */
    std::cout << "COORD: Receive a message.\n";
    Message incoming;
    Channel worker;
		my_communicator -> receive ( &incoming, 
                                 &worker, 
                                 my_communicator -> ANYTAG, 
                                 my_communicator -> ANYSOURCE );
		/* Deal with READY. */
		if ( incoming . tag == READY ) {
      std::cout << "COORD: Deal with READY.\n";
      Message job_message;
      /* Prepare a job. (if we cannot, break )*/
      int prepare_status = my_process -> prepare ( job_message );
			/* Check the status of prepare and act accordingly. */
      if ( prepare_status == 0 )  {
				job_message . tag = JOB;
        ++ pending;
        my_communicator -> send ( job_message, worker );
			} else {
        retiring = true;
      }
		} /* if */
		
		/* Deal with a RESULTS message */
		if ( incoming . tag == RESULTS ) {
      std::cout << "COORD: Deal with RESULTS.\n";
      -- pending;
      my_process -> accept ( incoming ); 
    }
  }
  
  // broadcast retire message
  std::cout << "COORD: Broadcast RETIRE.\n";
  Message retire_message;
  my_process -> prepare ( retire_message );
  retire_message . tag = RETIRE;
  my_communicator -> broadcast ( retire_message );
  
  std::cout << "COORD: Finalize.\n";
  my_process -> finalize ();
}

/* Coordinator_Worker_Process::run_worker() */
inline void Coordinator_Worker_Scheme::run_worker ( Coordinator_Worker_Process * my_process,
                                                    Communicator * my_communicator ) {
  /* Main loop */
  Channel boss = my_communicator -> DIRECTOR;
  while ( true ) {
    /* Send a READY message. */
    //std::cout << "Worker_Base::run -- send a READY message.\n";
#ifdef USEIDLEWORKERS
    if ( my_communicator -> SELF <= NUMBER_OF_WORKERS ) {
#endif
    Message ready;
    ready . tag = READY;
    my_communicator -> send ( ready, boss );
#ifdef USEIDLEWORKERS
    }
#endif
    /* Receive a job. */
    //std::cout << "Worker_Base::run -- receive a job\n";
    Message job_message;
		my_communicator -> receive ( &job_message, &boss, 
                                 my_communicator -> ANYTAG, my_communicator -> ANYSOURCE );
    /* Quit if it is a retire job. */
    if ( job_message . tag == RETIRE ) {
      //std::cout << "Worker_Base::run -- retiring\n";
      break;
    }
    /* Work the job. */
    //std::cout << "Worker_Base::run -- work a job\n";
    Message result_message;
    my_process -> work ( result_message, job_message );
		result_message . tag = RESULTS;
    /* Send the results back to the coordinator. */
    //std::cout << "Worker_Base::run -- report results\n";
    my_communicator -> send ( result_message, boss );
  }
}
{'fraction_non_alphanumeric': 0.1403760929452629, 'fraction_numerical': 0.003114145406635525, 'mean_word_length': 2.3121777072590244, 'pattern_counts': {'":': 0, '<': 50, '<?xml version=': 0, '>': 46, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3426391', 'n_tokens_mistral': 2487, 'n_tokens_neox': 2349, 'n_words': 627}