Merge branch 'atomic-ref'

Adds support for GCC's __atomic* built-ins and improves the performance
of logging (for ignored log levels) and half-open IKE_SA checking under
high loads.

Also fixes two potential race conditions in the load-tester plugin.
This commit is contained in:
Tobias Brunner
2014-04-24 18:00:11 +02:00
7 changed files with 136 additions and 31 deletions
+25 -7
View File
@@ -667,21 +667,39 @@ AC_COMPILE_IFELSE(
[AC_MSG_RESULT([no])] [AC_MSG_RESULT([no])]
) )
AC_MSG_CHECKING([for gcc atomic operations]) AC_MSG_CHECKING([for GCC __atomic operations])
AC_RUN_IFELSE([AC_LANG_SOURCE( AC_RUN_IFELSE([AC_LANG_SOURCE(
[[ [[
int main() { int main() {
volatile int ref = 1; int ref = 1, val;
__sync_fetch_and_add (&ref, 1); __atomic_fetch_add(&ref, 1, __ATOMIC_RELAXED);
__sync_sub_and_fetch (&ref, 1); val = __atomic_sub_fetch(&ref, 1, __ATOMIC_RELAXED);
/* Make sure test fails if operations are not supported */ __atomic_compare_exchange_n(&ref, &val, 0, 0, __ATOMIC_RELAXED,
__sync_val_compare_and_swap(&ref, 1, 0); __ATOMIC_RELAXED);
return ref; return ref;
} }
]])], ]])],
[AC_MSG_RESULT([yes]); [AC_MSG_RESULT([yes]);
AC_DEFINE([HAVE_GCC_ATOMIC_OPERATIONS], [], AC_DEFINE([HAVE_GCC_ATOMIC_OPERATIONS], [],
[have GCC __sync_* atomic operations])], [have GCC __atomic_* operations])],
[AC_MSG_RESULT([no])],
[AC_MSG_RESULT([no])]
)
AC_MSG_CHECKING([for GCC __sync operations])
AC_RUN_IFELSE([AC_LANG_SOURCE(
[[
int main() {
int ref = 1;
__sync_fetch_and_add (&ref, 1);
__sync_sub_and_fetch (&ref, 1);
__sync_val_compare_and_swap(&ref, 1, 0);
return ref;
}
]])],
[AC_MSG_RESULT([yes]);
AC_DEFINE([HAVE_GCC_SYNC_OPERATIONS], [],
[have GCC __sync_* operations])],
[AC_MSG_RESULT([no])], [AC_MSG_RESULT([no])],
[AC_MSG_RESULT([no])] [AC_MSG_RESULT([no])]
) )
+53 -7
View File
@@ -1,5 +1,5 @@
/* /*
* Copyright (C) 2011-2012 Tobias Brunner * Copyright (C) 2011-2014 Tobias Brunner
* Copyright (C) 2006 Martin Willi * Copyright (C) 2006 Martin Willi
* Hochschule fuer Technik Rapperswil * Hochschule fuer Technik Rapperswil
* *
@@ -23,6 +23,31 @@
#include <threading/mutex.h> #include <threading/mutex.h>
#include <threading/rwlock.h> #include <threading/rwlock.h>
/**
* These operations allow us to speed up the log level checks on some platforms.
* In particular if acquiring the read lock is expensive even in the absence of
* any writers.
*
* Note that while holding the read/write lock the read does not have to be
* atomic as the write lock must be held to set the level.
*/
#ifdef HAVE_GCC_ATOMIC_OPERATIONS
#define skip_level(ptr, level) (__atomic_load_n(ptr, __ATOMIC_RELAXED) < level)
#define set_level(ptr, val) __atomic_store_n(ptr, val, __ATOMIC_RELAXED)
#elif defined(HAVE_GCC_SYNC_OPERATIONS)
#define skip_level(ptr, level) (__sync_fetch_and_add(ptr, 0) < level)
#define set_level(ptr, val) __sync_bool_compare_and_swap(ptr, *ptr, val)
#else
#define skip_level(ptr, level) FALSE
#define set_level(ptr, val) ({ *ptr = val; })
#endif
typedef struct private_bus_t private_bus_t; typedef struct private_bus_t private_bus_t;
/** /**
@@ -173,11 +198,12 @@ static inline void register_logger(private_bus_t *this, debug_t group,
if (entry->logger->log) if (entry->logger->log)
{ {
this->max_level[group] = max(this->max_level[group], level); set_level(&this->max_level[group], max(this->max_level[group], level));
} }
if (entry->logger->vlog) if (entry->logger->vlog)
{ {
this->max_vlevel[group] = max(this->max_vlevel[group], level); set_level(&this->max_vlevel[group],
max(this->max_vlevel[group], level));
} }
} }
@@ -205,6 +231,7 @@ static inline void unregister_logger(private_bus_t *this, logger_t *logger)
if (found) if (found)
{ {
level_t level = LEVEL_SILENT, vlevel = LEVEL_SILENT;
debug_t group; debug_t group;
for (group = 0; group < DBG_MAX; group++) for (group = 0; group < DBG_MAX; group++)
@@ -214,13 +241,19 @@ static inline void unregister_logger(private_bus_t *this, logger_t *logger)
loggers = this->loggers[group]; loggers = this->loggers[group];
loggers->remove(loggers, found, NULL); loggers->remove(loggers, found, NULL);
this->max_level[group] = LEVEL_SILENT;
this->max_vlevel[group] = LEVEL_SILENT;
if (loggers->get_first(loggers, (void**)&entry) == SUCCESS) if (loggers->get_first(loggers, (void**)&entry) == SUCCESS)
{ {
this->max_level[group] = entry->levels[group]; if (entry->logger->log)
this->max_vlevel[group] = entry->levels[group]; {
level = entry->levels[group];
}
if (entry->logger->vlog)
{
vlevel = entry->levels[group];
}
} }
set_level(&this->max_level[group], level);
set_level(&this->max_vlevel[group], vlevel);
} }
} }
free(found); free(found);
@@ -324,6 +357,19 @@ METHOD(bus_t, vlog, void,
linked_list_t *loggers; linked_list_t *loggers;
log_data_t data; log_data_t data;
/* NOTE: This is not 100% thread-safe and done here only because it is
* performance critical. We therefore ignore the following two issues for
* this particular case: 1) We might miss some log messages if another
* thread concurrently increases the log level or registers a new logger.
* 2) We might have to acquire the read lock below even if it wouldn't be
* necessary anymore due to another thread concurrently unregistering a
* logger or reducing the level. */
if (skip_level(&this->max_level[group], level) &&
skip_level(&this->max_vlevel[group], level))
{
return;
}
this->log_lock->read_lock(this->log_lock); this->log_lock->read_lock(this->log_lock);
loggers = this->loggers[group]; loggers = this->loggers[group];
@@ -150,7 +150,7 @@ struct private_load_tester_config_t {
/** /**
* incremental numbering of generated configs * incremental numbering of generated configs
*/ */
u_int num; refcount_t num;
/** /**
* Dynamic source port, if used * Dynamic source port, if used
@@ -802,7 +802,7 @@ METHOD(backend_t, get_peer_cfg_by_name, peer_cfg_t*,
{ {
if (streq(name, "load-test")) if (streq(name, "load-test"))
{ {
return generate_config(this, this->num++); return generate_config(this, (u_int)ref_get(&this->num));
} }
return NULL; return NULL;
} }
@@ -31,14 +31,14 @@ struct private_load_tester_ipsec_t {
/** /**
* faked SPI counter * faked SPI counter
*/ */
u_int32_t spi; refcount_t spi;
}; };
METHOD(kernel_ipsec_t, get_spi, status_t, METHOD(kernel_ipsec_t, get_spi, status_t,
private_load_tester_ipsec_t *this, host_t *src, host_t *dst, private_load_tester_ipsec_t *this, host_t *src, host_t *dst,
u_int8_t protocol, u_int32_t reqid, u_int32_t *spi) u_int8_t protocol, u_int32_t reqid, u_int32_t *spi)
{ {
*spi = ++this->spi; *spi = (uint32_t)ref_get(&this->spi);
return SUCCESS; return SUCCESS;
} }
+8 -7
View File
@@ -353,6 +353,11 @@ struct private_ike_sa_manager_t {
*/ */
shareable_segment_t *half_open_segments; shareable_segment_t *half_open_segments;
/**
* Total number of half-open IKE_SAs.
*/
refcount_t half_open_count;
/** /**
* Hash table with connected_peers_t objects. * Hash table with connected_peers_t objects.
*/ */
@@ -764,6 +769,7 @@ static void put_half_open(private_ike_sa_manager_t *this, entry_t *entry)
this->half_open_table[row] = item; this->half_open_table[row] = item;
} }
this->half_open_segments[segment].count++; this->half_open_segments[segment].count++;
ref_get(&this->half_open_count);
lock->unlock(lock); lock->unlock(lock);
} }
@@ -803,6 +809,7 @@ static void remove_half_open(private_ike_sa_manager_t *this, entry_t *entry)
free(item); free(item);
} }
this->half_open_segments[segment].count--; this->half_open_segments[segment].count--;
ignore_result(ref_put(&this->half_open_count));
break; break;
} }
prev = item; prev = item;
@@ -1962,13 +1969,7 @@ METHOD(ike_sa_manager_t, get_half_open_count, u_int,
} }
else else
{ {
for (segment = 0; segment < this->segment_count; segment++) count = (u_int)ref_cur(&this->half_open_count);
{
lock = this->half_open_segments[segment].lock;
lock->read_lock(lock);
count += this->half_open_segments[segment].count;
lock->unlock(lock);
}
} }
return count; return count;
} }
+15 -3
View File
@@ -511,7 +511,7 @@ void nop()
{ {
} }
#ifndef HAVE_GCC_ATOMIC_OPERATIONS #if !defined(HAVE_GCC_ATOMIC_OPERATIONS) && !defined(HAVE_GCC_SYNC_OPERATIONS)
/** /**
* We use a single mutex for all refcount variables. * We use a single mutex for all refcount variables.
@@ -528,7 +528,6 @@ refcount_t ref_get(refcount_t *ref)
pthread_mutex_lock(&ref_mutex); pthread_mutex_lock(&ref_mutex);
current = ++(*ref); current = ++(*ref);
pthread_mutex_unlock(&ref_mutex); pthread_mutex_unlock(&ref_mutex);
return current; return current;
} }
@@ -545,6 +544,19 @@ bool ref_put(refcount_t *ref)
return !more_refs; return !more_refs;
} }
/**
* Current refcount
*/
refcount_t ref_cur(refcount_t *ref)
{
refcount_t current;
pthread_mutex_lock(&ref_mutex);
current = *ref;
pthread_mutex_unlock(&ref_mutex);
return current;
}
/** /**
* Single mutex for all compare and swap operations. * Single mutex for all compare and swap operations.
*/ */
@@ -566,7 +578,7 @@ bool cas_##name(type *ptr, type oldval, type newval) \
_cas_impl(bool, bool) _cas_impl(bool, bool)
_cas_impl(ptr, void*) _cas_impl(ptr, void*)
#endif /* HAVE_GCC_ATOMIC_OPERATIONS */ #endif /* !HAVE_GCC_ATOMIC_OPERATIONS && !HAVE_GCC_SYNC_OPERATIONS */
#ifdef HAVE_FMEMOPEN_FALLBACK #ifdef HAVE_FMEMOPEN_FALLBACK
+31 -3
View File
@@ -750,20 +750,40 @@ typedef u_int refcount_t;
#ifdef HAVE_GCC_ATOMIC_OPERATIONS #ifdef HAVE_GCC_ATOMIC_OPERATIONS
#define ref_get(ref) __atomic_add_fetch(ref, 1, __ATOMIC_RELAXED)
/* The relaxed memory model works fine for increments as these (usually) don't
* change the state of refcounted objects. But here we have to ensure that we
* free the right stuff if ref counted objects are mutable. So we have to sync
* with other threads that call ref_put(). It would be sufficient to use
* __ATOMIC_RELEASE here and then call __atomic_thread_fence() with
* __ATOMIC_ACQUIRE if we reach 0, but since we don't have control over the use
* of ref_put() we have to make sure. */
#define ref_put(ref) (!__atomic_sub_fetch(ref, 1, __ATOMIC_ACQ_REL))
#define ref_cur(ref) __atomic_load_n(ref, __ATOMIC_RELAXED)
#define _cas_impl(ptr, oldval, newval) ({ typeof(oldval) _old = oldval; \
__atomic_compare_exchange_n(ptr, &_old, newval, FALSE, \
__ATOMIC_SEQ_CST, __ATOMIC_RELAXED); })
#define cas_bool(ptr, oldval, newval) _cas_impl(ptr, oldval, newval)
#define cas_ptr(ptr, oldval, newval) _cas_impl(ptr, oldval, newval)
#elif defined(HAVE_GCC_SYNC_OPERATIONS)
#define ref_get(ref) __sync_add_and_fetch(ref, 1) #define ref_get(ref) __sync_add_and_fetch(ref, 1)
#define ref_put(ref) (!__sync_sub_and_fetch(ref, 1)) #define ref_put(ref) (!__sync_sub_and_fetch(ref, 1))
#define ref_cur(ref) __sync_fetch_and_add(ref, 0)
#define cas_bool(ptr, oldval, newval) \ #define cas_bool(ptr, oldval, newval) \
(__sync_bool_compare_and_swap(ptr, oldval, newval)) (__sync_bool_compare_and_swap(ptr, oldval, newval))
#define cas_ptr(ptr, oldval, newval) \ #define cas_ptr(ptr, oldval, newval) \
(__sync_bool_compare_and_swap(ptr, oldval, newval)) (__sync_bool_compare_and_swap(ptr, oldval, newval))
#else /* !HAVE_GCC_ATOMIC_OPERATIONS */ #else /* !HAVE_GCC_ATOMIC_OPERATIONS && !HAVE_GCC_SYNC_OPERATIONS */
/** /**
* Get a new reference. * Get a new reference.
* *
* Increments the reference counter atomic. * Increments the reference counter atomically.
* *
* @param ref pointer to ref counter * @param ref pointer to ref counter
* @return new value of ref * @return new value of ref
@@ -773,7 +793,7 @@ refcount_t ref_get(refcount_t *ref);
/** /**
* Put back a unused reference. * Put back a unused reference.
* *
* Decrements the reference counter atomic and * Decrements the reference counter atomically and
* says if more references available. * says if more references available.
* *
* @param ref pointer to ref counter * @param ref pointer to ref counter
@@ -781,6 +801,14 @@ refcount_t ref_get(refcount_t *ref);
*/ */
bool ref_put(refcount_t *ref); bool ref_put(refcount_t *ref);
/**
* Get the current value of the reference counter.
*
* @param ref pointer to ref counter
* @return current value of ref
*/
refcount_t ref_cur(refcount_t *ref);
/** /**
* Atomically replace value of ptr with newval if it currently equals oldval. * Atomically replace value of ptr with newval if it currently equals oldval.
* *