fixed_queue.c 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256
  1. /******************************************************************************
  2. *
  3. * Copyright (C) 2014 Google, Inc.
  4. *
  5. * Licensed under the Apache License, Version 2.0 (the "License");
  6. * you may not use this file except in compliance with the License.
  7. * You may obtain a copy of the License at:
  8. *
  9. * http://www.apache.org/licenses/LICENSE-2.0
  10. *
  11. * Unless required by applicable law or agreed to in writing, software
  12. * distributed under the License is distributed on an "AS IS" BASIS,
  13. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14. * See the License for the specific language governing permissions and
  15. * limitations under the License.
  16. *
  17. ******************************************************************************/
  18. #include "osi/allocator.h"
  19. #include "osi/fixed_queue.h"
  20. #include "osi/list.h"
  21. #include "osi/osi.h"
  22. #include "osi/mutex.h"
  23. #include "osi/semaphore.h"
  24. typedef struct fixed_queue_t {
  25. list_t *list;
  26. osi_sem_t enqueue_sem;
  27. osi_sem_t dequeue_sem;
  28. osi_mutex_t lock;
  29. size_t capacity;
  30. fixed_queue_cb dequeue_ready;
  31. } fixed_queue_t;
  32. fixed_queue_t *fixed_queue_new(size_t capacity)
  33. {
  34. fixed_queue_t *ret = osi_calloc(sizeof(fixed_queue_t));
  35. if (!ret) {
  36. goto error;
  37. }
  38. osi_mutex_new(&ret->lock);
  39. ret->capacity = capacity;
  40. ret->list = list_new(NULL);
  41. if (!ret->list) {
  42. goto error;
  43. }
  44. osi_sem_new(&ret->enqueue_sem, capacity, capacity);
  45. if (!ret->enqueue_sem) {
  46. goto error;
  47. }
  48. osi_sem_new(&ret->dequeue_sem, capacity, 0);
  49. if (!ret->dequeue_sem) {
  50. goto error;
  51. }
  52. return ret;
  53. error:;
  54. fixed_queue_free(ret, NULL);
  55. return NULL;
  56. }
  57. void fixed_queue_free(fixed_queue_t *queue, fixed_queue_free_cb free_cb)
  58. {
  59. const list_node_t *node;
  60. if (queue == NULL) {
  61. return;
  62. }
  63. fixed_queue_unregister_dequeue(queue);
  64. if (free_cb) {
  65. for (node = list_begin(queue->list); node != list_end(queue->list); node = list_next(node)) {
  66. free_cb(list_node(node));
  67. }
  68. }
  69. list_free(queue->list);
  70. osi_sem_free(&queue->enqueue_sem);
  71. osi_sem_free(&queue->dequeue_sem);
  72. osi_mutex_free(&queue->lock);
  73. osi_free(queue);
  74. }
  75. bool fixed_queue_is_empty(fixed_queue_t *queue)
  76. {
  77. bool is_empty = false;
  78. if (queue == NULL) {
  79. return true;
  80. }
  81. osi_mutex_lock(&queue->lock, OSI_MUTEX_MAX_TIMEOUT);
  82. is_empty = list_is_empty(queue->list);
  83. osi_mutex_unlock(&queue->lock);
  84. return is_empty;
  85. }
  86. size_t fixed_queue_length(fixed_queue_t *queue)
  87. {
  88. size_t length;
  89. if (queue == NULL) {
  90. return 0;
  91. }
  92. osi_mutex_lock(&queue->lock, OSI_MUTEX_MAX_TIMEOUT);
  93. length = list_length(queue->list);
  94. osi_mutex_unlock(&queue->lock);
  95. return length;
  96. }
  97. size_t fixed_queue_capacity(fixed_queue_t *queue)
  98. {
  99. assert(queue != NULL);
  100. return queue->capacity;
  101. }
  102. bool fixed_queue_enqueue(fixed_queue_t *queue, void *data, uint32_t timeout)
  103. {
  104. bool status=false; //Flag whether enqueued success
  105. assert(queue != NULL);
  106. assert(data != NULL);
  107. if (osi_sem_take(&queue->enqueue_sem, timeout) != 0) {
  108. return false;
  109. }
  110. osi_mutex_lock(&queue->lock, OSI_MUTEX_MAX_TIMEOUT);
  111. status = list_append(queue->list, data); //Check whether enqueued success
  112. osi_mutex_unlock(&queue->lock);
  113. if(status == true )
  114. osi_sem_give(&queue->dequeue_sem);
  115. return status;
  116. }
  117. void *fixed_queue_dequeue(fixed_queue_t *queue, uint32_t timeout)
  118. {
  119. void *ret = NULL;
  120. assert(queue != NULL);
  121. if (osi_sem_take(&queue->dequeue_sem, timeout) != 0) {
  122. return NULL;
  123. }
  124. osi_mutex_lock(&queue->lock, OSI_MUTEX_MAX_TIMEOUT);
  125. ret = list_front(queue->list);
  126. list_remove(queue->list, ret);
  127. osi_mutex_unlock(&queue->lock);
  128. osi_sem_give(&queue->enqueue_sem);
  129. return ret;
  130. }
  131. void *fixed_queue_try_peek_first(fixed_queue_t *queue)
  132. {
  133. void *ret = NULL;
  134. if (queue == NULL) {
  135. return NULL;
  136. }
  137. osi_mutex_lock(&queue->lock, OSI_MUTEX_MAX_TIMEOUT);
  138. ret = list_is_empty(queue->list) ? NULL : list_front(queue->list);
  139. osi_mutex_unlock(&queue->lock);
  140. return ret;
  141. }
  142. void *fixed_queue_try_peek_last(fixed_queue_t *queue)
  143. {
  144. void *ret = NULL;
  145. if (queue == NULL) {
  146. return NULL;
  147. }
  148. osi_mutex_lock(&queue->lock, OSI_MUTEX_MAX_TIMEOUT);
  149. ret = list_is_empty(queue->list) ? NULL : list_back(queue->list);
  150. osi_mutex_unlock(&queue->lock);
  151. return ret;
  152. }
  153. void *fixed_queue_try_remove_from_queue(fixed_queue_t *queue, void *data)
  154. {
  155. bool removed = false;
  156. if (queue == NULL) {
  157. return NULL;
  158. }
  159. osi_mutex_lock(&queue->lock, OSI_MUTEX_MAX_TIMEOUT);
  160. if (list_contains(queue->list, data) &&
  161. osi_sem_take(&queue->dequeue_sem, 0) == 0) {
  162. removed = list_remove(queue->list, data);
  163. assert(removed);
  164. }
  165. osi_mutex_unlock(&queue->lock);
  166. if (removed) {
  167. osi_sem_give(&queue->enqueue_sem);
  168. return data;
  169. }
  170. return NULL;
  171. }
  172. list_t *fixed_queue_get_list(fixed_queue_t *queue)
  173. {
  174. assert(queue != NULL);
  175. // NOTE: This function is not thread safe, and there is no point for
  176. // calling osi_mutex_lock() / osi_mutex_unlock()
  177. return queue->list;
  178. }
  179. void fixed_queue_register_dequeue(fixed_queue_t *queue, fixed_queue_cb ready_cb)
  180. {
  181. assert(queue != NULL);
  182. assert(ready_cb != NULL);
  183. queue->dequeue_ready = ready_cb;
  184. }
  185. void fixed_queue_unregister_dequeue(fixed_queue_t *queue)
  186. {
  187. assert(queue != NULL);
  188. queue->dequeue_ready = NULL;
  189. }
  190. void fixed_queue_process(fixed_queue_t *queue)
  191. {
  192. assert(queue != NULL);
  193. if (queue->dequeue_ready) {
  194. queue->dequeue_ready(queue);
  195. }
  196. }