<commit_before>#include "con_thread.hpp"

#include <unordered_map>

#include <cstdint>
#include <cstdio>
#include <cstddef>

#include <errno.h>
#include <sys/epoll.h>
#include <sys/types.h>
#include <unistd.h>

#include <boost/coroutine/all.hpp>

#include "io_utils.hpp"

using namespace std;
namespace co = boost::coroutines;

enum connection_co_status
{
	STATUS_ERROR = -1,
	STATUS_DONE = 0,
	STATUS_WAIT_READ,
	STATUS_WAIT_WRITE
};

template <typename YieldType>
static io_status read_n_co(YieldType& yield, int fd, void *buf, std::size_t count, std::size_t& count_out)
{
	io_status result;
	while((result = read_n(fd, buf, count, count_out)) == IO_AGAIN && count_out == 0)
	{
		yield(STATUS_WAIT_READ);
	}
	return result;
}

template <typename YieldType>
static io_status write_n_co(YieldType& yield, int fd, const void *buf, std::size_t count, std::size_t& count_out)
{
	io_status result;
	while((result = write_n(fd, buf, count, count_out)) == IO_AGAIN && count_out == 0)
	{
		yield(STATUS_WAIT_WRITE);
	}
	return result;
}

void echo_connection_co_func(co::symmetric_coroutine<connection_co_status>::yield_type& yield)
{
	auto read_yielder =
	[
}

void con_thread_func(int epoll_fd, const std::atomic_bool& run)
{
	struct con_state
	{
		std::array<char, 512> data;
		std::size_t size;
		bool writable;

		con_state(): size(0), writable(false) {}
	};
	std::unordered_map<int, con_state> con_data;<commit_msg>Coroutines<commit_after>#include "con_thread.hpp"

#include <unordered_map>

#include <cstdint>
#include <cstdio>
#include <cstddef>

#include <errno.h>
#include <sys/epoll.h>
#include <sys/types.h>
#include <unistd.h>

#include <boost/coroutine/all.hpp>

#include "io_utils.hpp"

using namespace std;
namespace co = boost::coroutines;

enum connection_co_status
{
	STATUS_ERROR = -1,
	STATUS_DONE = 0,
	STATUS_WAIT_READ,
	STATUS_WAIT_WRITE
};

template <typename YieldType>
static io_status read_n_co(YieldType& yield, int fd, void *buf, std::size_t count, std::size_t& count_out)
{
	io_status result;
	while((result = read_n(fd, buf, count, count_out)) == IO_AGAIN && count_out == 0)
	{
		yield(STATUS_WAIT_READ);
	}
	return result;
}

template <typename YieldType>
static io_status write_n_co(YieldType& yield, int fd, const void *buf, std::size_t count, std::size_t& count_out)
{
	io_status result;
	while((result = write_n(fd, buf, count, count_out)) == IO_AGAIN && count_out == 0)
	{
		yield(STATUS_WAIT_WRITE);
	}
	return result;
}

using connection_coroutine = co::symmetric_coroutine<void>;
using poller_coroutine = co::symmetric_coroutine<connection_co_status>;

struct connection_data
{
	int fd;
	connection_co_status status;
};

static void connection_co_func(connection_coroutine::yield_type& yield, poller_coroutine& poller, connection_data& data)
{
	auto yielder = bind(yield, ref(poller), placeholders::_1);

	char buffer[256];
	io_status result;
	while(true)
	{
		size_t count;
		result = read_n_co(yielder, data.fd, buffer, sizeof(buffer), count);
		if(result == IO_FAIL)
			break;
		result = write_n_co(yielder, data.fd, buffer, sizeof(buffer), count);
		if(result == IO_FAIL)
			break;
	}

	yield(poller, STATUS_DONE);
}

struct connection
{
	connection_coroutine co;
	connection_data data;
}

static void poller_co_func(poller_coroutine::yield_type& yield, poller_coroutine& this_co, int epoll_fd, atomic_bool& should_run)
{
	map<int, connection> cons;
	while(should_run.load())
	{
		struct epoll_event events[8];
		int num_events = epoll_wait(epoll_fd, events, 8, 50);
		for(int i = 0; i < num_events; ++i)
		{
			int fd = events[i].data.fd;
		    uint32_t event_mask = events[i].events;
			auto it = cons.find(fd);
			if(it == cons.end())
			{
				it = cons.emplace(fd, connection{});
				it->second.co = bind(connection_co_func, placeholders::_1, ref(this_co), ref(it->second));
				it->second.data = {fd, STATUS_WAIT_READ};
			}

			connection_co_status status = it->second.data.status;
			if((status == STATUS_WAIT_READ && (event_mask & EPOLLIN)) || (status == STATUS_WAIT_WRITE && (event_mask & EPOLLOUT)))
			{
				yield(it->second.co);
				status = yield.get();
			}

			if(status == STATUS_DONE || status == STATUS_FAIL)
			{
				goto end_con;
			}

			continue;
			
		end_con:
			close(fd);
			cons.erase(it);
		}
	}
}

void con_thread_func(int epoll_fd, const std::atomic_bool& run)
{
	struct con_state
	{
		std::array<char, 512> data;
		std::size_t size;
		bool writable;

		con_state(): size(0), writable(false) {}
	};
	std::unordered_map<int, con_state> con_data;
{'fraction_non_alphanumeric': 0.13674645390070922, 'fraction_numerical': 0.0062056737588652485, 'mean_word_length': 5.182191780821918, 'pattern_counts': {'":': 0, '<': 34, '<?xml version=': 0, '>': 38, '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': '862162', 'n_tokens_mistral': 1851, 'n_tokens_neox': 1740, 'n_words': 365}