partial lockless send for buffered plaintext

1 files 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;