<commit_before>#include "agenda.h"
#include "errors.h"
#include <unistd.h>
#include <thread>
#include <sstream>
#include <unistd.h>
#include <boost/bind.hpp>
#include <time.h>
#include <iostream>

using namespace es3;

agenda::agenda(size_t num_unbound, size_t num_cpu_bound, size_t num_io_bound,
			   bool quiet, bool final_quiet,
			   size_t segment_size, size_t max_segments_in_flight) :
	class_limits_ {{taskUnbound, num_unbound},
				   {taskCPUBound, num_cpu_bound},
				   {taskIOBound, num_io_bound}},
	quiet_(quiet), final_quiet_(final_quiet), segment_size_(segment_size),
	max_segments_in_flight_(max_segments_in_flight),
	num_working_(), num_submitted_(), num_done_(), num_failed_(),
	segments_in_flight_()
{
	clock_gettime(CLOCK_MONOTONIC, &start_time_) | libc_die2("Can't get time");
}

namespace es3
{
	struct segment_deleter
	{
		agenda_ptr parent_;

		void operator()(segment *seg)
		{
			delete seg;

			u_guard_t guard(parent_->m_);
			assert(parent_->segments_in_flight_>0);
			parent_->segments_in_flight_--;
			parent_->condition_.notify_one();
		}
	};

	class task_executor
	{
		agenda_ptr agenda_;
	public:
		task_executor(agenda_ptr agenda) : agenda_(agenda) {}

		std::pair<sync_task_ptr, std::vector<segment_ptr> > claim_task()
		{
			std::pair<sync_task_ptr, std::vector<segment_ptr> > res_pair;
			while(true)
			{
				u_guard_t lock(agenda_->m_);
				if (agenda_->tasks_.empty())
				{
					if (agenda_->num_working_==0)
						return res_pair;
					agenda_->condition_.wait(lock);
					continue;
				}

				//Iterate over classes to find one that is not yet full
				for(auto iter=agenda_->classes_.begin();
					iter!=agenda_->classes_.end();++iter)
				{
					//Check if there are too many tasks of this type running
					task_type_e cur_class=iter->first;

					size_t cur_num=iter->second;
					size_t limit=agenda_->get_capability(cur_class);
					//Unbound tasks are allowed to exceed their limits and
					//borrow threads from other classes
//					if (cur_class!=taskUnbound && limit<=cur_num)
//						continue; //Too busy
					if (limit<=cur_num)
						continue;

					size_t segments_avail=agenda_->max_segments_in_flight_-
							agenda_->segments_in_flight_;

					//Good! We can work on this class.
					//Find the task with the greatest segment requirements
					agenda::size_map_t::iterator pair=
							--agenda_->tasks_.end();
					size_t segments_needed=pair->first;
					if (segments_needed>segments_avail)
					{
						//Try the task with the least number of required segs
						printf("\nSegs bad %d\n\n", segments_needed);
						pair=agenda_->tasks_.begin();
						segments_needed=pair->first;
						printf("Segs good %d\n", segments_needed);
						if (segments_needed>segments_avail)
							continue; //No such luck :(
					}

					if (!pair->second.count(cur_class))
						continue; //No tasks for this class

					agenda::task_map_t &task_map=pair->second.at(cur_class);
					assert(!task_map.empty());
					sync_task_ptr res=task_map.begin()->second;
					task_map.erase(task_map.begin());
					if (task_map.empty())
					{
						pair->second.erase(cur_class);
						if (pair->second.empty())
							agenda_->tasks_.erase(pair->first);
					}

					agenda_->num_working_++;
					agenda_->classes_[cur_class]++;

					res_pair.first=res;
					if (segments_needed)
						res_pair.second=agenda_->get_segments(segments_needed);
					return res_pair;
				}

				agenda_->condition_.wait(lock);
			}<commit_msg>Remove tracing<commit_after>#include "agenda.h"
#include "errors.h"
#include <unistd.h>
#include <thread>
#include <sstream>
#include <unistd.h>
#include <boost/bind.hpp>
#include <time.h>
#include <iostream>

using namespace es3;

agenda::agenda(size_t num_unbound, size_t num_cpu_bound, size_t num_io_bound,
			   bool quiet, bool final_quiet,
			   size_t segment_size, size_t max_segments_in_flight) :
	class_limits_ {{taskUnbound, num_unbound},
				   {taskCPUBound, num_cpu_bound},
				   {taskIOBound, num_io_bound}},
	quiet_(quiet), final_quiet_(final_quiet), segment_size_(segment_size),
	max_segments_in_flight_(max_segments_in_flight),
	num_working_(), num_submitted_(), num_done_(), num_failed_(),
	segments_in_flight_()
{
	clock_gettime(CLOCK_MONOTONIC, &start_time_) | libc_die2("Can't get time");
}

namespace es3
{
	struct segment_deleter
	{
		agenda_ptr parent_;

		void operator()(segment *seg)
		{
			delete seg;

			u_guard_t guard(parent_->m_);
			assert(parent_->segments_in_flight_>0);
			parent_->segments_in_flight_--;
			parent_->condition_.notify_one();
		}
	};

	class task_executor
	{
		agenda_ptr agenda_;
	public:
		task_executor(agenda_ptr agenda) : agenda_(agenda) {}

		std::pair<sync_task_ptr, std::vector<segment_ptr> > claim_task()
		{
			std::pair<sync_task_ptr, std::vector<segment_ptr> > res_pair;
			while(true)
			{
				u_guard_t lock(agenda_->m_);
				if (agenda_->tasks_.empty())
				{
					if (agenda_->num_working_==0)
						return res_pair;
					agenda_->condition_.wait(lock);
					continue;
				}

				//Iterate over classes to find one that is not yet full
				for(auto iter=agenda_->classes_.begin();
					iter!=agenda_->classes_.end();++iter)
				{
					//Check if there are too many tasks of this type running
					task_type_e cur_class=iter->first;

					size_t cur_num=iter->second;
					size_t limit=agenda_->get_capability(cur_class);
					//Unbound tasks are allowed to exceed their limits and
					//borrow threads from other classes
//					if (cur_class!=taskUnbound && limit<=cur_num)
//						continue; //Too busy
					if (limit<=cur_num)
						continue;

					size_t segments_avail=agenda_->max_segments_in_flight_-
							agenda_->segments_in_flight_;

					//Good! We can work on this class.
					//Find the task with the greatest segment requirements
					agenda::size_map_t::iterator pair=
							--agenda_->tasks_.end();
					size_t segments_needed=pair->first;
					if (segments_needed>segments_avail)
					{
						//Try the task with the least number of required segs
						pair=agenda_->tasks_.begin();
						segments_needed=pair->first;
						if (segments_needed>segments_avail)
							continue; //No such luck :(
					}

					if (!pair->second.count(cur_class))
						continue; //No tasks for this class

					agenda::task_map_t &task_map=pair->second.at(cur_class);
					assert(!task_map.empty());
					sync_task_ptr res=task_map.begin()->second;
					task_map.erase(task_map.begin());
					if (task_map.empty())
					{
						pair->second.erase(cur_class);
						if (pair->second.empty())
							agenda_->tasks_.erase(pair->first);
					}

					agenda_->num_working_++;
					agenda_->classes_[cur_class]++;

					res_pair.first=res;
					if (segments_needed)
						res_pair.second=agenda_->get_segments(segments_needed);
					return res_pair;
				}

				agenda_->condition_.wait(lock);
			}
{'fraction_non_alphanumeric': 0.1259888661002051, 'fraction_numerical': 0.001464986815118664, 'mean_word_length': 3.6066126855600538, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 91, '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': '1623377', 'n_tokens_mistral': 2979, 'n_tokens_neox': 2345, 'n_words': 455}