<commit_before>#include "global_cached_private.h"

static void __complete_req(io_request *orig, thread_safe_page *p)
{
	int page_off = orig->get_offset() - ROUND_PAGE(orig->get_offset());

	p->lock();
	if (orig->get_access_method() == WRITE) {
		unsigned long *l = (unsigned long *) ((char *) p->get_data()
				+ page_off);
		unsigned long start = orig->get_offset() / sizeof(long);
		if (*l != start)
			printf("write: start: %ld, l: %ld, offset: %ld\n",
					start, *l, p->get_offset() / sizeof(long));
		memcpy((char *) p->get_data() + page_off, orig->get_buf(),
				orig->get_size());
		p->set_dirty(true);
	}
	else 
		/* I assume the data I read never crosses the page boundary */
		memcpy(orig->get_buf(), (char *) p->get_data() + page_off,
				orig->get_size());
	p->unlock();
	p->dec_ref();
}

class read_page_callback: public callback
{
public:
	int invoke(io_request *request) {
		io_request *orig = (io_request *) request->get_priv();
		thread_safe_page *p = (thread_safe_page *) orig->get_priv();
		std::vector<io_request> reqs;

		p->lock();
		p->set_data_ready(true);
		p->set_io_pending(false);
		io_request *req = p->get_io_req();
		assert(req);
		// TODO vector may allocate memory and block the thread.
		// I should be careful of that.
		while (req) {
			reqs.push_back(*req);
			req = req->get_next_req();
		}
		p->unlock();
		// At this point, data is ready, so any threads that want to add
		// requests to the page will fail.
		p->reset_reqs();

		for (unsigned i = 0; i < reqs.size(); i++) {
			__complete_req(&reqs[i], p);
			io_interface *io = reqs[i].get_io();
			if (io->get_callback())
				io->get_callback()->invoke(&reqs[i]);
		}
		return 0;
	}
};

global_cached_io::global_cached_io(io_interface *underlying,
		long cache_size, int cache_type) {
	cb = NULL;
	cache_hits = 0;
	this->underlying = underlying;
	underlying->set_callback(new read_page_callback());
	num_waits = 0;
	this->cache_size = cache_size;
	if (global_cache == NULL) {
		global_cache = create_cache(cache_type, cache_size);
	}
}

ssize_t global_cached_io::access(io_request *requests, int num)
{
	for (int i = 0; i < num; i++) {
		ssize_t ret;
		off_t old_off = -1;
		off_t offset = requests[i].get_offset();
		thread_safe_page *p = (thread_safe_page *) (get_global_cache()
				->search(ROUND_PAGE(offset), old_off));

		/*
		 * the page isn't in the cache,
		 * so the cache evict a page and return it to us.
		 */
		if (old_off != ROUND_PAGE(offset) && old_off != -1) {
			// TODO handle dirty pages later.
		}

		p->lock();
		if (!p->data_ready()) {
			if(!p->is_io_pending()) {
				p->set_io_pending(true);
				assert(!p->is_dirty());
				p->unlock();

				// If the thread can't add the request as the first IO request
				// of the page, it doesn't need to issue a IO request to the file.
				int status;
				io_request *orig = p->add_first_io_req(requests[i], status);
				if (status == ADD_REQ_SUCCESS) {
					io_request req((char *) p->get_data(),
							ROUND_PAGE(offset), PAGE_SIZE, READ,
							/*
							 * it will notify the underlying IO,
							 * which then notifies global_cached_io.
							 */
							underlying, (void *) orig);
					ret = underlying->access(&req, 1);
					if (ret < 0) {
						perror("read");
						exit(1);
					}
				}
				// If data is ready, the request won't be added to the page.
				// so just serve the data.
				else if (status == ADD_REQ_DATA_READY)
					goto serve_data;
			}
			else {
				p->unlock();
				// An IO request can be added only when the page is still
				// in IO pending state. Otherwise, it returns NULL.
				io_request *orig = p->add_io_req(requests[i]);
				// the page isn't in IO pending state any more.
				if (orig == NULL)
					goto serve_data;
			}
		}
		else {
			// If the data in the page is ready, we don't need to change any state
			// of the page and just read data.
			p->unlock();
serve_data:
			__complete_req(&requests[i], p);
			if (get_callback())
				get_callback()->invoke(&requests[i]);
#ifdef STATISTICS
			cache_hits++;
#endif
		}
	}
	return 0;
}

ssize_t global_cached_io::access(char *buf, off_t offset,
		ssize_t size, int access_method)
{
	ssize_t ret;
	off_t old_off = -1;
	thread_safe_page *p = (thread_safe_page *) (get_global_cache()
			->search(ROUND_PAGE(offset), old_off));

	/*<commit_msg>calculate cache hit rate correctly in global cache IO.<commit_after>#include "global_cached_private.h"

static void __complete_req(io_request *orig, thread_safe_page *p)
{
	int page_off = orig->get_offset() - ROUND_PAGE(orig->get_offset());

	p->lock();
	if (orig->get_access_method() == WRITE) {
		unsigned long *l = (unsigned long *) ((char *) p->get_data()
				+ page_off);
		unsigned long start = orig->get_offset() / sizeof(long);
		if (*l != start)
			printf("write: start: %ld, l: %ld, offset: %ld\n",
					start, *l, p->get_offset() / sizeof(long));
		memcpy((char *) p->get_data() + page_off, orig->get_buf(),
				orig->get_size());
		p->set_dirty(true);
	}
	else 
		/* I assume the data I read never crosses the page boundary */
		memcpy(orig->get_buf(), (char *) p->get_data() + page_off,
				orig->get_size());
	p->unlock();
	p->dec_ref();
}

class read_page_callback: public callback
{
public:
	int invoke(io_request *request) {
		io_request *orig = (io_request *) request->get_priv();
		thread_safe_page *p = (thread_safe_page *) orig->get_priv();
		std::vector<io_request> reqs;

		p->lock();
		p->set_data_ready(true);
		p->set_io_pending(false);
		io_request *req = p->get_io_req();
		assert(req);
		// TODO vector may allocate memory and block the thread.
		// I should be careful of that.
		while (req) {
			reqs.push_back(*req);
			req = req->get_next_req();
		}
		p->unlock();
		// At this point, data is ready, so any threads that want to add
		// requests to the page will fail.
		p->reset_reqs();

		for (unsigned i = 0; i < reqs.size(); i++) {
			__complete_req(&reqs[i], p);
			io_interface *io = reqs[i].get_io();
			if (io->get_callback())
				io->get_callback()->invoke(&reqs[i]);
		}
		return 0;
	}
};

global_cached_io::global_cached_io(io_interface *underlying,
		long cache_size, int cache_type) {
	cb = NULL;
	cache_hits = 0;
	this->underlying = underlying;
	underlying->set_callback(new read_page_callback());
	num_waits = 0;
	this->cache_size = cache_size;
	if (global_cache == NULL) {
		global_cache = create_cache(cache_type, cache_size);
	}
}

ssize_t global_cached_io::access(io_request *requests, int num)
{
	for (int i = 0; i < num; i++) {
		ssize_t ret;
		off_t old_off = -1;
		off_t offset = requests[i].get_offset();
		thread_safe_page *p = (thread_safe_page *) (get_global_cache()
				->search(ROUND_PAGE(offset), old_off));

		/*
		 * the page isn't in the cache,
		 * so the cache evict a page and return it to us.
		 */
		if (old_off != ROUND_PAGE(offset) && old_off != -1) {
			// TODO handle dirty pages later.
		}

		p->lock();
		if (!p->data_ready()) {
			if(!p->is_io_pending()) {
				p->set_io_pending(true);
				assert(!p->is_dirty());
				p->unlock();

				// If the thread can't add the request as the first IO request
				// of the page, it doesn't need to issue a IO request to the file.
				int status;
				io_request *orig = p->add_first_io_req(requests[i], status);
				if (status == ADD_REQ_SUCCESS) {
					io_request req((char *) p->get_data(),
							ROUND_PAGE(offset), PAGE_SIZE, READ,
							/*
							 * it will notify the underlying IO,
							 * which then notifies global_cached_io.
							 */
							underlying, (void *) orig);
					ret = underlying->access(&req, 1);
					if (ret < 0) {
						perror("read");
						exit(1);
					}
				}
				else {
#ifdef STATISTICS
					cache_hits++;
#endif
					// If data is ready, the request won't be added to the page.
					// so just serve the data.
					if (status == ADD_REQ_DATA_READY)
						goto serve_data;
				}
			}
			else {
				p->unlock();
#ifdef STATISTICS
				cache_hits++;
#endif
				// An IO request can be added only when the page is still
				// in IO pending state. Otherwise, it returns NULL.
				io_request *orig = p->add_io_req(requests[i]);
				// the page isn't in IO pending state any more.
				if (orig == NULL)
					goto serve_data;
			}
		}
		else {
			// If the data in the page is ready, we don't need to change any state
			// of the page and just read data.
			p->unlock();
#ifdef STATISTICS
			cache_hits++;
#endif
serve_data:
			__complete_req(&requests[i], p);
			if (get_callback())
				get_callback()->invoke(&requests[i]);
		}
	}
	return 0;
}

ssize_t global_cached_io::access(char *buf, off_t offset,
		ssize_t size, int access_method)
{
	ssize_t ret;
	off_t old_off = -1;
	thread_safe_page *p = (thread_safe_page *) (get_global_cache()
			->search(ROUND_PAGE(offset), old_off));

	/*
{'fraction_non_alphanumeric': 0.15043538038496793, 'fraction_numerical': 0.002749770852428964, 'mean_word_length': 3.2684596577017113, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 97, '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': '5416695', 'n_tokens_mistral': 3627, 'n_tokens_neox': 3099, 'n_words': 846}