text: #define _GNU_SOURCE
  #include <features.h>
  
  #include <assert.h>
  #include <proton/types.h>
  #include <pthread.h>
  #include <stdio.h>
  #include <stdlib.h>
  #include <time.h>
  
  #include "rb.h"
  #include "utils.h"
  
  rb_rwbytes_t *rb_alloc(int count, int buf_size, bool wake_producer) {
      rb_rwbytes_t *rb = malloc(sizeof(rb_rwbytes_t));
  
      rb->count = count;
      rb->buf_size = buf_size;
      rb->wake_producer = wake_producer;
  
      if ((rb->ring_buffer = malloc(count * sizeof(pn_rwbytes_t))) == NULL) {
          free(rb);
  
          return NULL;
      }
  
      for (int i = 0; i < count; i++) {
          if ((rb->ring_buffer[i].start = malloc(buf_size)) == NULL) {
              rb_free(rb);
  
              return NULL;
          }
          rb->ring_buffer[i].size = 0;
      }
      rb->head = 0;
      rb->tail = count - 1;
  
      rb->overruns = 0;
      rb->processed = 0;
      rb->queue_block = 0;
  
      rb->total_active.tv_sec = 0;
      rb->total_active.tv_nsec = 0;
  
      rb->total_wait.tv_sec = 0;
      rb->total_wait.tv_nsec = 0;
  
      rb->total_t1.tv_sec = 0;
      rb->total_t1.tv_nsec = 0;
  
      rb->total_t2.tv_sec = 0;
      rb->total_t2.tv_nsec = 0;
  
      pthread_cond_init(&rb->rb_ready, NULL);
      pthread_mutex_init(&rb->rb_mutex, NULL);
  
      if (rb->wake_producer) {
          pthread_cond_init(&rb->rb_free, NULL);
      }
  
      return rb;
  }
  
  void rb_free(rb_rwbytes_t *rb) {
      if (rb == NULL) {
          return;
      }
      for (int i = 0; i < rb->count; i++) {
          free(rb->ring_buffer[i].start);
      }
      free(rb->ring_buffer);
      free(rb);
  }
  
  pn_rwbytes_t *rb_get_head(rb_rwbytes_t *rb) {
      if (rb == NULL) {
          return NULL;
      }
      return &rb->ring_buffer[rb->head];
  }
  
  pn_rwbytes_t *rb_get_tail(rb_rwbytes_t *rb) {
      if (rb == NULL) {
          return NULL;
      }
      return &rb->ring_buffer[rb->tail];
  }
  
  // Place the already allocated buffer entry in the
  // queue.  The producer does not block as it needs
  // to continually process the incoming AMQP messagaes.
  // Just need to wake up the consumer if it is waiting
  // for messages.
  pn_rwbytes_t *rb_put(rb_rwbytes_t *rb) {
      if (rb == NULL) {
          return NULL;
      }
      pn_rwbytes_t *next_buffer = NULL;
  
      int next = (rb->head + 1) % rb->count;
      if (next != rb->tail) {
          rb->head = next;
          next_buffer = &rb->ring_buffer[rb->head];
          pthread_mutex_lock(&rb->rb_mutex);
          pthread_cond_broadcast(&rb->rb_ready);
          pthread_mutex_unlock(&rb->rb_mutex);
      } else {
          rb->overruns++;
          rb->ring_buffer[rb->head].size = 0;
      }
  
      return next_buffer; // May be NULL
  }
  
  pn_rwbytes_t *rb_get(rb_rwbytes_t *rb) {
      if (rb == NULL) {
          return NULL;
      }
  
      int next;
  
      next = (rb->tail + 1) % rb->count;
      while (next == rb->head) {
          pthread_mutex_lock(&rb->rb_mutex);
          pthread_cond_wait(&rb->rb_ready, &rb->rb_mutex);
          pthread_mutex_unlock(&rb->rb_mutex);
  
          next = (rb->tail + 1) % rb->count;
          rb->queue_block++;
      }
      // set data size to zero
      rb->ring_buffer[rb->tail].size = 0;
  
      rb->tail = next;
  
      if (rb->wake_producer && rb_free_size(rb) == 1) {
          pthread_mutex_lock(&rb->rb_mutex);
          pthread_cond_broadcast(&rb->rb_free);
          pthread_mutex_unlock(&rb->rb_mutex);
      }
  
      rb->processed++;
  
      return &rb->ring_buffer[rb->tail];
  }
  
  int rb_inuse_size(rb_rwbytes_t *rb) { return rb->count - rb_free_size(rb); }
  
  int rb_free_size(rb_rwbytes_t *rb) {
      assert(rb->head != rb->tail);
  
      int head = rb->head;
      int tail = rb->tail;
  
      return head > tail ? rb->count - (head - tail) - 1 : tail - head - 1;
  }
  
  int rb_size(rb_rwbytes_t *rb) { return rb->count; }
  
  long rb_get_overruns(rb_rwbytes_t *rb) { return rb->overruns; }
  
  long rb_get_processed(rb_rwbytes_t *rb) { return rb->processed; }
  
  long rb_get_queue_block(rb_rwbytes_t *rb) { return rb->queue_block; }
  