From ea4d0f1360c2c23764decea94efcc38a8e47cafb Mon Sep 17 00:00:00 2001 From: Test_User Date: Thu, 20 Jun 2024 02:38:25 -0400 Subject: partial lockless send for buffered plaintext --- networks/plaintext_buffered.c | 57 ++++++++++++++++++++++++++++++++++++------- 1 file changed, 48 insertions(+), 9 deletions(-) diff --git a/networks/plaintext_buffered.c b/networks/plaintext_buffered.c index eb53555..437c8fa 100644 --- a/networks/plaintext_buffered.c +++ b/networks/plaintext_buffered.c @@ -75,25 +75,40 @@ void * plaintext_buffered_send_thread(void *handle) { size_t read_buffer_index = 0; ssize_t res = 0; while (1) { +#ifdef USE_FUTEX +#else mutex_lock(&(info->mutex)); +#endif size_t len; +#ifdef USE_FUTEX + len = __sync_sub_and_fetch(&(info->buffer_len), res); +#else info->buffer_len -= (size_t)res; len = info->buffer_len; +#endif #ifdef USE_FUTEX mutex_unlock(&(info->release_write)); - if (len == 0) - info->release_read = 1; #else sem_trywait(&(info->release_write)); sem_post(&(info->release_write)); #endif +#ifdef USE_FUTEX + if (!__sync_fetch_and_or(&(info->valid), 0)) { // TODO: Clean up mutexes in exit code too + mutex_lock(&(info->mutex)); + goto plaintext_buffered_send_thread_error_unlock; + } +#else if (!info->valid) goto plaintext_buffered_send_thread_error_unlock; +#endif +#ifdef USE_FUTEX +#else mutex_unlock(&(info->mutex)); +#endif #ifdef USE_FUTEX if (len == 0) { @@ -171,20 +186,36 @@ void * plaintext_buffered_send_thread(void *handle) { int plaintext_buffered_send(void *handle, struct string msg) { struct plaintext_buffered_handle *plaintext_handle = handle; while (msg.len > 0) { - size_t len = msg.len; +#ifdef USE_FUTEX +#else + mutex_lock(&(plaintext_handle->mutex)); +#endif + +#ifdef USE_FUTEX + size_t len = PLAINTEXT_BUFFERED_LEN - __sync_fetch_and_or(&(plaintext_handle->buffer_len), 0); // There's no fetch-only for __sync +#else + size_t len = PLAINTEXT_BUFFERED_LEN - plaintext_handle->buffer_len; +#endif + + if (len > msg.len) + len = msg.len; + if (len > PLAINTEXT_BUFFERED_LEN - plaintext_handle->write_buffer_index) len = PLAINTEXT_BUFFERED_LEN - plaintext_handle->write_buffer_index; - mutex_lock(&(plaintext_handle->mutex)); + +#ifdef USE_FUTEX + if (!__sync_fetch_and_or(&(plaintext_handle->valid), 0)) { + return 1; + } +#else if (!plaintext_handle->valid) { mutex_unlock(&(plaintext_handle->mutex)); return 1; } - if (len > PLAINTEXT_BUFFERED_LEN - plaintext_handle->buffer_len) - len = PLAINTEXT_BUFFERED_LEN - plaintext_handle->buffer_len; +#endif + #ifdef USE_FUTEX if (len == 0) { - plaintext_handle->release_write = 1; - mutex_unlock(&(plaintext_handle->mutex)); mutex_lock(&(plaintext_handle->release_write)); continue; } @@ -196,14 +227,22 @@ int plaintext_buffered_send(void *handle, struct string msg) { } #endif memcpy(&(plaintext_handle->buffer[plaintext_handle->write_buffer_index]), msg.data, len); + +#ifdef USE_FUTEX + __sync_fetch_and_add(&(plaintext_handle->buffer_len), len); // No __sync add-only either +#else plaintext_handle->buffer_len += len; +#endif + #ifdef USE_FUTEX mutex_unlock(&(plaintext_handle->release_read)); #else + mutex_unlock(&(plaintext_handle->mutex)); + sem_trywait(&(plaintext_handle->release_read)); sem_post(&(plaintext_handle->release_read)); #endif - mutex_unlock(&(plaintext_handle->mutex)); + plaintext_handle->write_buffer_index += len; if (plaintext_handle->write_buffer_index >= PLAINTEXT_BUFFERED_LEN) plaintext_handle->write_buffer_index = 0; -- cgit v1.2.3