ringbuffer_queue.c 4.04 KB
Newer Older
1 2 3 4 5
/*
 * 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
6
 * the OAI Public License, Version 1.1  (the "License"); you may not use this file
7 8 9 10 11 12 13 14 15 16 17 18 19 20
 * 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
 */ 
21 22 23 24 25 26 27 28 29 30 31

/*! \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"
32
#include "common/utils/LOG/log.h"
33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69

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;

70 71 72
  if (size <= 0)
    goto error;

73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103
  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:
  LOG_E(MAC, "%s: an error occured\n", __FUNCTION__);
  return -1;
}

104
int message_get(message_queue_t *queue, void **data, int *priority) {
105 106 107 108 109 110 111 112 113 114 115
  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;
116
  const int size = m->size;
117 118
  *priority = m->priority;
  free(m);
119
  return size;
120 121
}

122 123 124 125 126
void message_get_unlock(message_queue_t *queue) {
  /* don't do anything, this function exists to unlock a message_queue but is
   * not needed in the case of the ringbuffer_queue */
}

shahab's avatar
shahab committed
127
void destroy_message_queue(message_queue_t *queue) {
128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143
  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);
}