<commit_before>#ifndef __PART_GLOBAL_CACHED_PRIVATE_H__
#define __PART_GLOBAL_CACHED_PRIVATE_H__

#include <errno.h>

#define BUF_SIZE 1000
#define REQ_QUEUE_SIZE 100000
#define REPLY_QUEUE_SIZE REQ_QUEUE_SIZE

#include "garbage_collection.h"
#include "global_cached_private.h"

class io_reply
{
	char *buf;
	off_t offset;
	ssize_t size: 32;
	int success: 1;
	int status: 16;
	int access_method: 1;
	void init(char *buf, off_t off, ssize_t size, int success,
			int status, int access_method) {
		this->buf = buf;
		this->offset = off;
		this->size = size;
		this->success = success;
		this->status = status;
		this->access_method = access_method;
	}
public:
	io_reply() {
		init(NULL, 0, 0, 0, 0, READ);
	}

	io_reply(io_request *req, int success, int status) {
		init(req->get_buf(), req->get_offset(), req->get_size(),
					success, status, req->get_access_method());
	}

	int get_status() {
		return status;
	}

	bool is_success() {
		return success;
	}

	char *get_buf() {
		return buf;
	}

	off_t get_offset() {
		return offset;
	}

	ssize_t get_size() {
		return size;
	}

	int get_access_method() {
		return access_method;
	}
};

class part_global_cached_private;

struct thread_group
{
	int id;
	int nthreads;
	part_global_cached_private **threads;
	page_cache *cache;
};

inline int min(int v1, int v2)
{
	return v1 > v2 ? v2 : v1;
}

/**
 * this is a thread-safe FIFO queue.
 * It supports bulk operations.
 */
template<class T>
class bulk_queue
{
	T *buf;
	volatile int size;
	int start;
	int num_entries;
	pthread_spinlock_t _lock;
public:
	bulk_queue(int size) {
		buf = new T[size];
		this->size = size;
		start = 0;
		num_entries = 0;
		pthread_spin_init(&_lock, PTHREAD_PROCESS_PRIVATE);
	}

	~bulk_queue() {
		pthread_spin_destroy(&_lock);
		delete [] buf;
	}

	int fetch(T *entries, int num) {
		pthread_spin_lock(&_lock);
		int n = min(num, num_entries);
		for (int i = 0; i < n; i++) {
			entries[i] = buf[(start + i) % this->size];
		}
		start = (start + n) % this->size;
		num_entries -= n;
		pthread_spin_unlock(&_lock);
		return n;
	}

	int add(T *entries, int num) {
		pthread_spin_lock(&_lock);
		int n = min(num, this->size - num_entries);
		int end = (start + num_entries) % this->size;
		for (int i = 0; i < n; i++) {
			buf[(end + i) % this->size] = entries[i];
		}
		num_entries += n;
		pthread_spin_unlock(&_lock);
		return n;
	}

	int get_num_entries() {
		return num_entries;
	}

	bool is_full() {
		return num_entries == size;
	}

	bool is_empty() {
		return num_entries == 0;
	}
};

class part_global_cached_private: public global_cached_private
{
	static thread_group *groups;
	/* this mutex just for helping initialize cache. */
	static pthread_mutex_t init_mutex;
	/* indicates the number of threads that finish initialization. */
	static int num_finish_init;
	static pthread_mutex_t wait_mutex;
	static pthread_cond_t cond;

	int num_groups;
	int group_idx;

	/* the size of the cache associated to the thread. */
	long cache_size;
	int cache_type;
	
	bulk_queue<io_request> *request_queue;
	bulk_queue<io_reply> *reply_queue;

	volatile int finished_threads;

	/* 
	 * we have request and reply buffer to distribute requests and replies
	 * in case we can't send them all.
	 */
	/* there is a request buffer for each group it sends to */
	io_request **thread_reqs;
	int *nreqs;
	/* there is a reply buffer for each thread it sends to */
	io_reply **thread_replies;
	int *nreplies;

	int hash_req(io_request *req)
	{
		return req->get_offset() % num_groups;
	}

	long processed_requests;

public:
	static int group_id(int thread_id, int num_groups) {
		/* number of threads in a group */
		int num_threads = nthreads / num_groups;
		return thread_id / num_threads;
	}
	/* get the location of a thread in the group. */
	static int thread_idx(int thread_id, int num_groups) {
		/* number of threads in a group */
		int num_threads = nthreads / num_groups;
		return thread_id % num_threads;
	}

	part_global_cached_private *id2thread(int thread_id) {
		return groups[group_id(thread_id, num_groups)].threads[thread_idx(thread_id, num_groups)];
	}

	~part_global_cached_private() {
		for (int i = 0; i < num_groups; i++) {
			numa_free(thread_reqs[i], sizeof(io_request) * BUF_SIZE);
			numa_free(thread_replies[i], sizeof(io_reply) * BUF_SIZE);
		}
		numa_free(thread_reqs, sizeof(thread_reqs[0]) * num_groups);
		numa_free(thread_replies, sizeof(thread_replies[0]) * num_groups);
		delete request_queue;
		delete reply_queue;
	}<commit_msg>fix a bug in computing the hash for requests.<commit_after>#ifndef __PART_GLOBAL_CACHED_PRIVATE_H__
#define __PART_GLOBAL_CACHED_PRIVATE_H__

#include <errno.h>

#define BUF_SIZE 1000
#define REQ_QUEUE_SIZE 100000
#define REPLY_QUEUE_SIZE REQ_QUEUE_SIZE

#include "garbage_collection.h"
#include "global_cached_private.h"

class io_reply
{
	char *buf;
	off_t offset;
	ssize_t size: 32;
	int success: 1;
	int status: 16;
	int access_method: 1;
	void init(char *buf, off_t off, ssize_t size, int success,
			int status, int access_method) {
		this->buf = buf;
		this->offset = off;
		this->size = size;
		this->success = success;
		this->status = status;
		this->access_method = access_method;
	}
public:
	io_reply() {
		init(NULL, 0, 0, 0, 0, READ);
	}

	io_reply(io_request *req, int success, int status) {
		init(req->get_buf(), req->get_offset(), req->get_size(),
					success, status, req->get_access_method());
	}

	int get_status() {
		return status;
	}

	bool is_success() {
		return success;
	}

	char *get_buf() {
		return buf;
	}

	off_t get_offset() {
		return offset;
	}

	ssize_t get_size() {
		return size;
	}

	int get_access_method() {
		return access_method;
	}
};

class part_global_cached_private;

struct thread_group
{
	int id;
	int nthreads;
	part_global_cached_private **threads;
	page_cache *cache;
};

inline int min(int v1, int v2)
{
	return v1 > v2 ? v2 : v1;
}

/**
 * this is a thread-safe FIFO queue.
 * It supports bulk operations.
 */
template<class T>
class bulk_queue
{
	T *buf;
	volatile int size;
	int start;
	int num_entries;
	pthread_spinlock_t _lock;
public:
	bulk_queue(int size) {
		buf = new T[size];
		this->size = size;
		start = 0;
		num_entries = 0;
		pthread_spin_init(&_lock, PTHREAD_PROCESS_PRIVATE);
	}

	~bulk_queue() {
		pthread_spin_destroy(&_lock);
		delete [] buf;
	}

	int fetch(T *entries, int num) {
		pthread_spin_lock(&_lock);
		int n = min(num, num_entries);
		for (int i = 0; i < n; i++) {
			entries[i] = buf[(start + i) % this->size];
		}
		start = (start + n) % this->size;
		num_entries -= n;
		pthread_spin_unlock(&_lock);
		return n;
	}

	int add(T *entries, int num) {
		pthread_spin_lock(&_lock);
		int n = min(num, this->size - num_entries);
		int end = (start + num_entries) % this->size;
		for (int i = 0; i < n; i++) {
			buf[(end + i) % this->size] = entries[i];
		}
		num_entries += n;
		pthread_spin_unlock(&_lock);
		return n;
	}

	int get_num_entries() {
		return num_entries;
	}

	bool is_full() {
		return num_entries == size;
	}

	bool is_empty() {
		return num_entries == 0;
	}
};

class part_global_cached_private: public global_cached_private
{
	static thread_group *groups;
	/* this mutex just for helping initialize cache. */
	static pthread_mutex_t init_mutex;
	/* indicates the number of threads that finish initialization. */
	static int num_finish_init;
	static pthread_mutex_t wait_mutex;
	static pthread_cond_t cond;

	int num_groups;
	int group_idx;

	/* the size of the cache associated to the thread. */
	long cache_size;
	int cache_type;
	
	bulk_queue<io_request> *request_queue;
	bulk_queue<io_reply> *reply_queue;

	volatile int finished_threads;

	/* 
	 * we have request and reply buffer to distribute requests and replies
	 * in case we can't send them all.
	 */
	/* there is a request buffer for each group it sends to */
	io_request **thread_reqs;
	int *nreqs;
	/* there is a reply buffer for each thread it sends to */
	io_reply **thread_replies;
	int *nreplies;

	int hash_req(io_request *req)
	{
		return req->get_offset() / get_entry_size() % num_groups;
	}

	long processed_requests;

public:
	static int group_id(int thread_id, int num_groups) {
		/* number of threads in a group */
		int num_threads = nthreads / num_groups;
		return thread_id / num_threads;
	}
	/* get the location of a thread in the group. */
	static int thread_idx(int thread_id, int num_groups) {
		/* number of threads in a group */
		int num_threads = nthreads / num_groups;
		return thread_id % num_threads;
	}

	part_global_cached_private *id2thread(int thread_id) {
		return groups[group_id(thread_id, num_groups)].threads[thread_idx(thread_id, num_groups)];
	}

	~part_global_cached_private() {
		for (int i = 0; i < num_groups; i++) {
			numa_free(thread_reqs[i], sizeof(io_request) * BUF_SIZE);
			numa_free(thread_replies[i], sizeof(io_reply) * BUF_SIZE);
		}
		numa_free(thread_reqs, sizeof(thread_reqs[0]) * num_groups);
		numa_free(thread_replies, sizeof(thread_replies[0]) * num_groups);
		delete request_queue;
		delete reply_queue;
	}
{'fraction_non_alphanumeric': 0.11677211677211677, 'fraction_numerical': 0.00777000777000777, 'mean_word_length': 4.087521174477696, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 47, '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': '1526830', 'n_tokens_mistral': 3606, 'n_tokens_neox': 3368, 'n_words': 891}