openairinterface5g/openair2/UTIL/ASYNC_IF/ringbuffer_queue.c

/*
 * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The OpenAirInterface Software Alliance licenses this file to You under
 * the OAI Public License, Version 1.1  (the "License"); you may not use this file
 * except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.openairinterface.org/?page_id=698
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *-------------------------------------------------------------------------------
 * For more information about the OpenAirInterface (OAI) Software Alliance:
 *      contact@openairinterface.org
 */ 

/*! \file ringbuffer_queue.c
 * \brief Lock-free ringbuffer used for async message passing of agent
 * \author Xenofon Foukas
 * \date March 2016
 * \version 1.0
 * \email: x.foukas@sms.ed.ac.uk
 * @ingroup _mac
 */

#include "ringbuffer_queue.h"
#include "common/utils/LOG/log.h"

message_queue_t * new_message_queue(int size) {
  
  message_queue_t *ret = NULL;

  ret = calloc(1, sizeof(message_queue_t));
  if (ret == NULL)
    goto error;

  lfds700_misc_library_init_valid_on_current_logical_core();
  lfds700_misc_prng_init(&(ret->ps));
  ret->ringbuffer_array = malloc(sizeof(struct lfds700_ringbuffer_element) * size);
  lfds700_ringbuffer_init_valid_on_current_logical_core(&(ret->ringbuffer_state),
							ret->ringbuffer_array,
							size,
							&(ret->ps),
							NULL);

  return ret;
  
 error:
  LOG_E(MAC, "%s: an error occured\n", __FUNCTION__);
  if (ret != NULL) {
    free(ret->ringbuffer_array);
    memset(ret, 0, sizeof(message_queue_t));
    free(ret);
  }
  return NULL;
}

int message_put(message_queue_t *queue, void *data, int size, int priority) {

  struct lfds700_misc_prng_state ls;
  enum lfds700_misc_flag overwrite_occurred_flag;
  message_t *overwritten_msg;
  message_t *m = NULL;

  LFDS700_MISC_MAKE_VALID_ON_CURRENT_LOGICAL_CORE_INITS_COMPLETED_BEFORE_NOW_ON_ANY_OTHER_LOGICAL_CORE;
  lfds700_misc_prng_init(&ls);
  
  m = calloc(1, sizeof(message_t));
  if (m == NULL)
    goto error;

  m->data = data;
  m->size = size;
  m->priority = priority;

  lfds700_ringbuffer_write(&(queue->ringbuffer_state),
			   NULL,
			   (void *) m,
			   &overwrite_occurred_flag,
			   NULL,
			   (void **) &overwritten_msg,
			   &ls);

  if (overwrite_occurred_flag == LFDS700_MISC_FLAG_RAISED) {
    free(overwritten_msg->data);
    free(overwritten_msg);
  }

  return 0;

 error:
  free(m);
  LOG_E(MAC, "%s: an error occured\n", __FUNCTION__);
  return -1;
}

int message_get(message_queue_t *queue, void **data, int *size, int *priority) {
  message_t *m;
  struct lfds700_misc_prng_state ls;
  
  LFDS700_MISC_MAKE_VALID_ON_CURRENT_LOGICAL_CORE_INITS_COMPLETED_BEFORE_NOW_ON_ANY_OTHER_LOGICAL_CORE;
  lfds700_misc_prng_init(&ls);

  if (lfds700_ringbuffer_read(&(queue->ringbuffer_state), NULL, (void **) &m, &ls) == 0) {
    return -1;
  }

  *data = m->data;
  *size = m->size;
  *priority = m->priority;
  free(m);
  return 0;
}

void destroy_message_queue(message_queue_t *queue) {
  struct lfds700_misc_prng_state ls;

  message_t *m;

  LFDS700_MISC_MAKE_VALID_ON_CURRENT_LOGICAL_CORE_INITS_COMPLETED_BEFORE_NOW_ON_ANY_OTHER_LOGICAL_CORE;
  lfds700_misc_prng_init(&ls);

  while (lfds700_ringbuffer_read(&(queue->ringbuffer_state), NULL, (void **) &m, &ls) != 0) {
    free(m->data);
    memset(m, 0, sizeof(message_t));
    free(m);
  }
  free(queue->ringbuffer_array);
  memset(queue, 0, sizeof(message_queue_t));
  free(queue);
}