Line data    Source code

       1             : #include "netio/netio.h"
       2             : 
       3             : /**
       4             :  * @brief Creates a stack of completion objects
       5             :  *
       6             :  * A message can be bound to multiple subscribers.
       7             :  * A completion object (co) is associated to each message.
       8             :  * The co is held until the message has been delivered to all the subscribers.
       9             :  * To keep track of how many subscribers received the message the co is equipped with a semaphore.
      10             :  * The semaphore has a threshold equal to the number of subscriptions and a counter.
      11             :  * When the counter exceeds the threshold the semaphore fires;
      12             :  * only then `msg_published` will be called and the DMA read pointer advanced.
      13             :  *
      14             :  * @param netio_completion_stack: stack of completion objects
      15             :  * @param num:     total number of completion objects
      16             :  */
      17             : 
      18             : void
      19          18 : netio_completion_stack_init(struct netio_completion_stack* stack, size_t num)
      20             : {
      21          18 :     stack->objects = malloc(num * sizeof(struct netio_completion_object));
      22          18 :     stack->stack = malloc(num * sizeof(struct netio_completion_object*));
      23          18 :     stack->key_array = malloc(num * sizeof(uint64_t));
      24        9234 :     for(size_t i=0; i<num; i++) {
      25        9216 :       stack->stack[i] = &(stack->objects[i]);
      26        9216 :       stack->stack[i]->key = 0xFFFFFFFFFFFFFFFF;
      27        9216 :       stack->key_array[i] = 0xFFFFFFFFFFFFFFFF;
      28             :     }
      29          18 :     stack->available_objects = num;
      30          18 :     stack->num_objects = num;
      31          18 :     stack->buf.data = stack->objects;
      32          18 :     stack->buf.size = num * sizeof(struct netio_completion_object);
      33          18 :     stack->printed = 0;
      34          18 : }
      35             : 
      36             : 
      37             : void
      38          17 : netio_completion_stack_register_send_socket(struct netio_completion_stack* stack, struct netio_send_socket* socket)
      39             : {
      40          17 :     netio_register_send_buffer(socket, &stack->buf, 0);
      41          17 : }
      42             : 
      43             : 
      44             : int
      45    18137380 : netio_completion_stack_pop(struct netio_completion_stack* stack, struct netio_completion_object** object)
      46             : {
      47    18137380 :     if(stack->available_objects == 0) {
      48             :       return 1;
      49             :     }
      50    10805972 :     stack->available_objects--;
      51    10805972 :     *object = stack->stack[stack->available_objects];
      52    10805972 :     return 0;
      53             : }
      54             : 
      55             : 
      56             : int
      57    10805972 : netio_completion_stack_push(struct netio_completion_stack* stack, struct netio_completion_object* object)
      58             : {
      59    10805972 :     if(stack->available_objects == stack->num_objects) {
      60             :       return 1;
      61             :     }
      62    10805972 :     object->key = 0xFFFFFFFFFFFFFFFF;
      63    10805972 :     stack->stack[stack->available_objects] = object;
      64    10805972 :     stack->key_array[stack->available_objects] = 0xFFFFFFFFFFFFFFFF;
      65    10805972 :     stack->available_objects++;
      66    10805972 :     if(stack->printed == 1 && stack->available_objects == 250){
      67           0 :       stack->printed = 0;
      68             :     }
      69             :     return 0;
      70             : }