<commit_before>/* This file is part of Ingen.
 * Copyright (C) 2008-2009 <NAME> <http://drobilla.net>
 *
 * Ingen is free software; you can redistribute it and/or modify it under the
 * terms of the GNU General Public License as published by the Free Software
 * Foundation; either version 2 of the License, or (at your option) any later
 * version.
 *
 * Ingen is distributed in the hope that it will be useful, but WITHOUT ANY
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE.  See the GNU General Public License for details.
 *
 * You should have received a copy of the GNU General Public License along
 * with this program; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
 */

#include <sys/mman.h>
#include "EventSource.hpp"
#include "QueuedEvent.hpp"
#include "PostProcessor.hpp"
#include "ThreadManager.hpp"
#include "ProcessContext.hpp"

using namespace std;

namespace Ingen {


EventSource::EventSource(size_t queue_size)
	: _blocking_semaphore(0)
{
	Thread::set_context(THREAD_PRE_PROCESS);
	set_name("EventSource");
}


EventSource::~EventSource()
{
	Thread::stop();
}


/** Push an unprepared event onto the queue.
 */
void
EventSource::push_queued(QueuedEvent* const ev)
{
	assert(!ev->is_prepared());
	Raul::List<Event*>::Node* node = new Raul::List<Event*>::Node(ev);
	_events.push_back(node);
	if (_prepared_back.get() == NULL)
		_prepared_back = node;

	whip();
}


/** Process all events for a cycle.
 *
 * Executed events will be pushed to @a dest.
 */
void
EventSource::process(PostProcessor& dest, ProcessContext& context, bool limit)
{
	ThreadManager::assert_thread(THREAD_PROCESS);

	if (_events.empty())
		return;

	/* Limit the maximum number of queued events to process per cycle.  This
	 * makes the process callback (more) realtime-safe by preventing being
	 * choked by events coming in faster than they can be processed.
	 * FIXME: test this and figure out a good value */
	const size_t MAX_QUEUED_EVENTS = context.nframes() / 32;

	size_t num_events_processed = 0;

	QueuedEvent*              ev       = (QueuedEvent*)_events.front();
	Raul::List<Event*>::Node* new_head = _events.head();

	while (ev && ev->is_prepared() && ev->time() < context.end()) {
		ev->execute(context);
		new_head = new_head->next();
		++num_events_processed;
		if (limit && num_events_processed > MAX_QUEUED_EVENTS)
			break;
		ev = (new_head ? (QueuedEvent*)new_head->elem() : NULL);
	}

	if (num_events_processed > 0) {
		Raul::List<Event*> front;
		_events.chop_front(front, num_events_processed, new_head);
		dest.append(&front);
	}
}


/** Pre-process a single event */<commit_msg>Fix race crash processing event list (Raul::ListImpl seriously needs an audit and heavy testing...)<commit_after>/* This file is part of Ingen.
 * Copyright (C) 2008-2009 <NAME> <http://drobilla.net>
 *
 * Ingen is free software; you can redistribute it and/or modify it under the
 * terms of the GNU General Public License as published by the Free Software
 * Foundation; either version 2 of the License, or (at your option) any later
 * version.
 *
 * Ingen is distributed in the hope that it will be useful, but WITHOUT ANY
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE.  See the GNU General Public License for details.
 *
 * You should have received a copy of the GNU General Public License along
 * with this program; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
 */

#include <sys/mman.h>
#include "EventSource.hpp"
#include "QueuedEvent.hpp"
#include "PostProcessor.hpp"
#include "ThreadManager.hpp"
#include "ProcessContext.hpp"

using namespace std;

namespace Ingen {


EventSource::EventSource(size_t queue_size)
	: _blocking_semaphore(0)
{
	Thread::set_context(THREAD_PRE_PROCESS);
	set_name("EventSource");
}


EventSource::~EventSource()
{
	Thread::stop();
}


/** Push an unprepared event onto the queue.
 */
void
EventSource::push_queued(QueuedEvent* const ev)
{
	assert(!ev->is_prepared());
	Raul::List<Event*>::Node* node = new Raul::List<Event*>::Node(ev);
	_events.push_back(node);
	if (_prepared_back.get() == NULL)
		_prepared_back = node;

	whip();
}


/** Process all events for a cycle.
 *
 * Executed events will be pushed to @a dest.
 */
void
EventSource::process(PostProcessor& dest, ProcessContext& context, bool limit)
{
	ThreadManager::assert_thread(THREAD_PROCESS);

	if (_events.empty())
		return;

	/* Limit the maximum number of queued events to process per cycle.  This
	 * makes the process callback (more) realtime-safe by preventing being
	 * choked by events coming in faster than they can be processed.
	 * FIXME: test this and figure out a good value */
	const size_t MAX_QUEUED_EVENTS = context.nframes() / 32;

	size_t num_events_processed = 0;

	Raul::List<Event*>::Node* head = _events.head();
	Raul::List<Event*>::Node* tail = head;

	if (!head)
		return;
	
	QueuedEvent* ev = (QueuedEvent*)head->elem();

	while (ev && ev->is_prepared() && ev->time() < context.end()) {
		ev->execute(context);
		tail = head;
		head = head->next();
		++num_events_processed;
		if (limit && num_events_processed > MAX_QUEUED_EVENTS)
			break;
		ev = (head ? (QueuedEvent*)head->elem() : NULL);
	}

	if (num_events_processed > 0) {
		Raul::List<Event*> front;
		_events.chop_front(front, num_events_processed, tail);
		dest.append(&front);
	}
}


/** Pre-process a single event */
{'fraction_non_alphanumeric': 0.12040374909877433, 'fraction_numerical': 0.009012256669069935, 'mean_word_length': 4.466995073891626, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 35, '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': '894088', 'n_tokens_mistral': 1848, 'n_tokens_neox': 1724, 'n_words': 581}