<commit_before>/*
 *
 *  D-Bus++ - C++ bindings for D-Bus
 *
 *  Copyright (C) 2005-2007  <NAME> <<EMAIL>>
 *
 *
 *  This library is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU Lesser General Public
 *  License as published by the Free Software Foundation; either
 *  version 2.1 of the License, or (at your option) any later version.
 *
 *  This library 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
 *  Lesser General Public License for more details.
 *
 *  You should have received a copy of the GNU Lesser General Public
 *  License along with this library; if not, write to the Free Software
 *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 *
 */

#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#include <dbus-c++/dispatcher.h>

#include <dbus/dbus.h>

#include "dispatcher_p.h"
#include "server_p.h"
#include "connection_p.h"

DBus::Dispatcher *DBus::default_dispatcher = NULL;

using namespace DBus;

Timeout::Timeout(Timeout::Internal *i)
: _int(i)
{
	dbus_timeout_set_data((DBusTimeout *)i, this, NULL);
}

int Timeout::interval() const
{
	return dbus_timeout_get_interval((DBusTimeout *)_int);
}

bool Timeout::enabled() const
{
	return dbus_timeout_get_enabled((DBusTimeout *)_int);
}

bool Timeout::handle()
{
	return dbus_timeout_handle((DBusTimeout *)_int);
}

/*
*/

Watch::Watch(Watch::Internal *i)
: _int(i)
{
	dbus_watch_set_data((DBusWatch *)i, this, NULL);
}

int Watch::descriptor() const
{
#if HAVE_WIN32
	return dbus_watch_get_socket((DBusWatch*)_int);
#else
	return dbus_watch_get_unix_fd((DBusWatch*)_int);
#endif
}

int Watch::flags() const
{
	return dbus_watch_get_flags((DBusWatch *)_int);
}

bool Watch::enabled() const
{
	return dbus_watch_get_enabled((DBusWatch *)_int);
}

bool Watch::handle(int flags)
{
	return dbus_watch_handle((DBusWatch *)_int, flags);
}

/*
*/

dbus_bool_t Dispatcher::Private::on_add_watch(DBusWatch *watch, void *data)
{
	Dispatcher *d = static_cast<Dispatcher *>(data);

	Watch::Internal *w = reinterpret_cast<Watch::Internal *>(watch);

	d->add_watch(w);

	return true;
}

void Dispatcher::Private::on_rem_watch(DBusWatch *watch, void *data)
{
	Dispatcher *d = static_cast<Dispatcher *>(data);

	Watch *w = static_cast<Watch *>(dbus_watch_get_data(watch));

	d->rem_watch(w);
}

void Dispatcher::Private::on_toggle_watch(DBusWatch *watch, void *data)
{
	Watch *w = static_cast<Watch *>(dbus_watch_get_data(watch));

	w->toggle();
}

dbus_bool_t Dispatcher::Private::on_add_timeout(DBusTimeout *timeout, void *data)
{
	Dispatcher *d = static_cast<Dispatcher *>(data);

	Timeout::Internal *t = reinterpret_cast<Timeout::Internal *>(timeout);

	d->add_timeout(t);

	return true;
}

void Dispatcher::Private::on_rem_timeout(DBusTimeout *timeout, void *data)
{
	Dispatcher *d = static_cast<Dispatcher *>(data);

	Timeout *t = static_cast<Timeout *>(dbus_timeout_get_data(timeout));

	d->rem_timeout(t);
}

void Dispatcher::Private::on_toggle_timeout(DBusTimeout *timeout, void *data)
{
	Timeout *t = static_cast<Timeout *>(dbus_timeout_get_data(timeout));

	t->toggle();
}

void Dispatcher::queue_connection(Connection::Private *cp)
{
	_mutex_p.lock();
	_pending_queue.push_back(cp);
	_mutex_p.unlock();
}


bool Dispatcher::has_something_to_dispatch()
{
	_mutex_p.lock();
	bool has_something = false;
	for(Connection::PrivatePList::iterator it = _pending_queue.begin();
		it != _pending_queue.end() && !has_something;
		++it)
	{
		has_something = (*it)->has_something_to_dispatch();
	}

	_mutex_p.unlock();
	return has_something;
}


void Dispatcher::dispatch_pending()
{
	_mutex_p.lock();

	// SEEME: dbus-glib is dispatching only one message at a time to not starve the loop/other things...

	while (_pending_queue.size() > 0)
	{
		Connection::PrivatePList::iterator i, j;
		
		i = _pending_queue.begin();

		while (i != _pending_queue.end())
		{
			j = i; 
			
			++j;

			if ((*i)->do_dispatch())
				_pending_queue.erase(i);

			i = j;
		}
	}
	_mutex_p.unlock();
}

void DBus::_init_threading()
{
#ifdef DBUS_HAS_THREADS_INIT_DEFAULT
	dbus_threads_init_default();
#else
	debug_log("Thread support is not enabled! Your D-Bus version is too old!");
#endif//DBUS_HAS_THREADS_INIT_DEFAULT
}

void DBus::_init_threading(
	MutexNewFn m1,
	MutexFreeFn m2,
	MutexLockFn m3,
	MutexUnlockFn m4,
	CondVarNewFn c1,
	CondVarFreeFn c2,
	CondVarWaitFn c3,
	CondVarWaitTimeoutFn c4,
	CondVarWakeOneFn c5,
	CondVarWakeAllFn c6
)<commit_msg>fix connection queue deadlock<commit_after>/*
 *
 *  D-Bus++ - C++ bindings for D-Bus
 *
 *  Copyright (C) 2005-2007  <NAME> <<EMAIL>>
 *
 *
 *  This library is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU Lesser General Public
 *  License as published by the Free Software Foundation; either
 *  version 2.1 of the License, or (at your option) any later version.
 *
 *  This library 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
 *  Lesser General Public License for more details.
 *
 *  You should have received a copy of the GNU Lesser General Public
 *  License along with this library; if not, write to the Free Software
 *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 *
 */

#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#include <dbus-c++/dispatcher.h>

#include <dbus/dbus.h>

#include "dispatcher_p.h"
#include "server_p.h"
#include "connection_p.h"

DBus::Dispatcher *DBus::default_dispatcher = NULL;

using namespace DBus;

Timeout::Timeout(Timeout::Internal *i)
: _int(i)
{
	dbus_timeout_set_data((DBusTimeout *)i, this, NULL);
}

int Timeout::interval() const
{
	return dbus_timeout_get_interval((DBusTimeout *)_int);
}

bool Timeout::enabled() const
{
	return dbus_timeout_get_enabled((DBusTimeout *)_int);
}

bool Timeout::handle()
{
	return dbus_timeout_handle((DBusTimeout *)_int);
}

/*
*/

Watch::Watch(Watch::Internal *i)
: _int(i)
{
	dbus_watch_set_data((DBusWatch *)i, this, NULL);
}

int Watch::descriptor() const
{
#if HAVE_WIN32
	return dbus_watch_get_socket((DBusWatch*)_int);
#else
	return dbus_watch_get_unix_fd((DBusWatch*)_int);
#endif
}

int Watch::flags() const
{
	return dbus_watch_get_flags((DBusWatch *)_int);
}

bool Watch::enabled() const
{
	return dbus_watch_get_enabled((DBusWatch *)_int);
}

bool Watch::handle(int flags)
{
	return dbus_watch_handle((DBusWatch *)_int, flags);
}

/*
*/

dbus_bool_t Dispatcher::Private::on_add_watch(DBusWatch *watch, void *data)
{
	Dispatcher *d = static_cast<Dispatcher *>(data);

	Watch::Internal *w = reinterpret_cast<Watch::Internal *>(watch);

	d->add_watch(w);

	return true;
}

void Dispatcher::Private::on_rem_watch(DBusWatch *watch, void *data)
{
	Dispatcher *d = static_cast<Dispatcher *>(data);

	Watch *w = static_cast<Watch *>(dbus_watch_get_data(watch));

	d->rem_watch(w);
}

void Dispatcher::Private::on_toggle_watch(DBusWatch *watch, void *data)
{
	Watch *w = static_cast<Watch *>(dbus_watch_get_data(watch));

	w->toggle();
}

dbus_bool_t Dispatcher::Private::on_add_timeout(DBusTimeout *timeout, void *data)
{
	Dispatcher *d = static_cast<Dispatcher *>(data);

	Timeout::Internal *t = reinterpret_cast<Timeout::Internal *>(timeout);

	d->add_timeout(t);

	return true;
}

void Dispatcher::Private::on_rem_timeout(DBusTimeout *timeout, void *data)
{
	Dispatcher *d = static_cast<Dispatcher *>(data);

	Timeout *t = static_cast<Timeout *>(dbus_timeout_get_data(timeout));

	d->rem_timeout(t);
}

void Dispatcher::Private::on_toggle_timeout(DBusTimeout *timeout, void *data)
{
	Timeout *t = static_cast<Timeout *>(dbus_timeout_get_data(timeout));

	t->toggle();
}

void Dispatcher::queue_connection(Connection::Private *cp)
{
	_mutex_p.lock();
	_pending_queue.push_back(cp);
	_mutex_p.unlock();
}


bool Dispatcher::has_something_to_dispatch()
{
	_mutex_p.lock();
	bool has_something = false;
	for(Connection::PrivatePList::iterator it = _pending_queue.begin();
		it != _pending_queue.end() && !has_something;
		++it)
	{
		has_something = (*it)->has_something_to_dispatch();
	}

	_mutex_p.unlock();
	return has_something;
}


void Dispatcher::dispatch_pending()
{
	while (1)
	{
		_mutex_p.lock();
	        bool queue_empty = _pending_queue.size() <= 0;
		_mutex_p.unlock();

		if ( queue_empty ) return;

		Connection::Private *entry = NULL;

		_mutex_p.lock();
		Connection::PrivatePList::iterator i =  _pending_queue.begin();
		if (i != _pending_queue.end())
		{
			entry = *i;
			_pending_queue.erase(i);
		}
		_mutex_p.unlock();

		if ( entry == NULL ) return;

		if ( entry->do_dispatch() ) // there is no more data
		{
			// delete entry;
			// A leak???
			return;
		}

		// There is more data - queue connection again
		queue_connection( entry );
	}
}

void DBus::_init_threading()
{
#ifdef DBUS_HAS_THREADS_INIT_DEFAULT
	dbus_threads_init_default();
#else
	debug_log("Thread support is not enabled! Your D-Bus version is too old!");
#endif//DBUS_HAS_THREADS_INIT_DEFAULT
}

void DBus::_init_threading(
	MutexNewFn m1,
	MutexFreeFn m2,
	MutexLockFn m3,
	MutexUnlockFn m4,
	CondVarNewFn c1,
	CondVarFreeFn c2,
	CondVarWaitFn c3,
	CondVarWaitTimeoutFn c4,
	CondVarWakeOneFn c5,
	CondVarWakeAllFn c6
)
{'fraction_non_alphanumeric': 0.13341831315062672, 'fraction_numerical': 0.007966857871255577, 'mean_word_length': 5.247511612475116, 'pattern_counts': {'":': 0, '<': 36, '<?xml version=': 0, '>': 52, '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': '6770701', 'n_tokens_mistral': 3525, 'n_tokens_neox': 3368, 'n_words': 800}