<commit_before>/*
 * Copyright 2014 Open Connectome Project (http://openconnecto.me)
 * Written by <NAME> (<EMAIL>)
 *
 * This file is part of FlashGraph.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <stdlib.h>
#include <malloc.h>

#include "safs_file.h"
#include "in_mem_storage.h"
#include "cache.h"

class in_mem_io: public io_interface
{
	const in_mem_graph &graph;
	int file_id;
	fifo_queue<io_request> req_buf;
	fifo_queue<user_compute *> compute_buf;
	fifo_queue<user_compute *> incomp_computes;

	void process_req(const io_request &req);
	void process_computes();
public:
	in_mem_io(const in_mem_graph &_graph, int file_id,
			thread *t): io_interface(t), graph(_graph), req_buf(
				get_node_id(), 1024), compute_buf(get_node_id(), 1024,
				true), incomp_computes(get_node_id(), 1024, true) {
		this->file_id = file_id;
	}

	virtual int get_file_id() const {
		return file_id;
	}

	virtual bool support_aio() {
		return true;
	}

	virtual void flush_requests() { }

	virtual int num_pending_ios() const {
		return 0;
	}

	virtual void access(io_request *requests, int num, io_status *status);
	virtual int wait4complete(int) {
		assert(compute_buf.is_empty());
		if (!incomp_computes.is_empty()) {
			compute_buf.add(&incomp_computes);
			assert(incomp_computes.is_empty());
			process_computes();
		}
		return 0;
	}
};

class in_mem_byte_array: public page_byte_array
{
	const io_request &req;
	thread_safe_page *pages;
public:
	in_mem_byte_array(const io_request &_req, thread_safe_page *pages): req(_req) {
		this->pages = pages;
	}

	virtual off_t get_offset_in_first_page() const {
		return req.get_offset() % PAGE_SIZE;
	}

	virtual thread_safe_page *get_page(int pg_idx) const {
		return &pages[pg_idx];
	}

	virtual size_t get_size() const {
		return req.get_size();
	}

	void lock() {
		assert(0);
	}

	void unlock() {
		assert(0);
	}
};

enum
{
	IN_QUEUE,
};

void in_mem_io::process_req(const io_request &req)
{
	assert(req.get_req_type() == io_request::USER_COMPUTE);
	in_mem_byte_array byte_arr(req, &graph.graph_pages[req.get_offset() / PAGE_SIZE]);
	user_compute *compute = req.get_compute();
	compute->run(byte_arr);
	// If the user compute hasn't completed and it's not in the queue,
	// add it to the queue.
	if (!compute->has_completed() && !compute->test_flag(IN_QUEUE)) {
		compute->set_flag(IN_QUEUE, true);
		compute_buf.push_back(compute);
	}
	else
		compute->dec_ref();

	if (compute->has_completed() && !compute->test_flag(IN_QUEUE)) {
		assert(compute->get_ref() == 0);
		// It might still be referenced by the graph engine.
		if (compute->get_ref() == 0) {
			compute_allocator *alloc = compute->get_allocator();<commit_msg>[Graph]: grow the queues in in-mem storage.<commit_after>/*
 * Copyright 2014 Open Connectome Project (http://openconnecto.me)
 * Written by <NAME> (<EMAIL>)
 *
 * This file is part of FlashGraph.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <stdlib.h>
#include <malloc.h>

#include "safs_file.h"
#include "in_mem_storage.h"
#include "cache.h"

class in_mem_io: public io_interface
{
	const in_mem_graph &graph;
	int file_id;
	fifo_queue<io_request> req_buf;
	fifo_queue<user_compute *> compute_buf;
	fifo_queue<user_compute *> incomp_computes;

	void process_req(const io_request &req);
	void process_computes();
public:
	in_mem_io(const in_mem_graph &_graph, int file_id,
			thread *t): io_interface(t), graph(_graph), req_buf(
				get_node_id(), 1024), compute_buf(get_node_id(), 1024,
				true), incomp_computes(get_node_id(), 1024, true) {
		this->file_id = file_id;
	}

	virtual int get_file_id() const {
		return file_id;
	}

	virtual bool support_aio() {
		return true;
	}

	virtual void flush_requests() { }

	virtual int num_pending_ios() const {
		return 0;
	}

	virtual void access(io_request *requests, int num, io_status *status);
	virtual int wait4complete(int) {
		assert(compute_buf.is_empty());
		if (!incomp_computes.is_empty()) {
			compute_buf.add(&incomp_computes);
			assert(incomp_computes.is_empty());
			process_computes();
		}
		return 0;
	}
};

class in_mem_byte_array: public page_byte_array
{
	const io_request &req;
	thread_safe_page *pages;
public:
	in_mem_byte_array(const io_request &_req, thread_safe_page *pages): req(_req) {
		this->pages = pages;
	}

	virtual off_t get_offset_in_first_page() const {
		return req.get_offset() % PAGE_SIZE;
	}

	virtual thread_safe_page *get_page(int pg_idx) const {
		return &pages[pg_idx];
	}

	virtual size_t get_size() const {
		return req.get_size();
	}

	void lock() {
		assert(0);
	}

	void unlock() {
		assert(0);
	}
};

enum
{
	IN_QUEUE,
};

void in_mem_io::process_req(const io_request &req)
{
	assert(req.get_req_type() == io_request::USER_COMPUTE);
	in_mem_byte_array byte_arr(req, &graph.graph_pages[req.get_offset() / PAGE_SIZE]);
	user_compute *compute = req.get_compute();
	compute->run(byte_arr);
	// If the user compute hasn't completed and it's not in the queue,
	// add it to the queue.
	if (!compute->has_completed() && !compute->test_flag(IN_QUEUE)) {
		compute->set_flag(IN_QUEUE, true);
		if (compute_buf.is_full())
			compute_buf.expand_queue(compute_buf.get_size() * 2);
		compute_buf.push_back(compute);
	}
	else
		compute->dec_ref();

	if (compute->has_completed() && !compute->test_flag(IN_QUEUE)) {
		assert(compute->get_ref() == 0);
		// It might still be referenced by the graph engine.
		if (compute->get_ref() == 0) {
			compute_allocator *alloc = compute->get_allocator();
{'fraction_non_alphanumeric': 0.12288266091777025, 'fraction_numerical': 0.008469356328919002, 'mean_word_length': 4.6774475524475525, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '6319715', 'n_tokens_mistral': 2425, 'n_tokens_neox': 2263, 'n_words': 591}