Line data    Source code

       1             : #include "netio/netio.h"
       2             : 
       3             : #include <string.h>
       4             : #include <stdio.h>
       5             : #include "log.h"
       6             : 
       7             : #if defined DEBUG || defined DEBUG_MR
       8             : #define __FILENAME__ (strrchr(__FILE__, '/') ? strrchr(__FILE__, '/') + 1 : __FILE__)
       9             : #define log_dbg(...) log_log(LOG_DEBUG, __FILENAME__, __LINE__, __VA_ARGS__)
      10             : #define log_trc(...) log_log(LOG_TRACE, __FILENAME__, __LINE__, __VA_ARGS__)
      11             : #else
      12             : #define log_dbg(...)
      13             : #define log_trc(...)
      14             : #endif
      15             : 
      16             : 
      17             : static void
      18          11 : init_bufferstack(struct netio_buffer_stack* stack, unsigned num, size_t bufsize)
      19             : {
      20          11 :   stack->buffers = malloc(num*sizeof(struct netio_buffer*));
      21          11 :   stack->buffer_addresses = malloc(num*sizeof(struct netio_buffer*));
      22          11 :   stack->num_buffers = num;
      23          11 :   stack->available_buffers = num;
      24        2339 :   for(unsigned int i=0; i<num; i++) {
      25        2328 :     stack->buffers[i] = malloc(sizeof(struct netio_buffer));
      26        2328 :     stack->buffers[i]->data = malloc(bufsize);
      27        2328 :     stack->buffers[i]->size = bufsize;
      28        2328 :     stack->buffers[i]->mr = NULL;
      29        2328 :     stack->buffer_addresses[i] = stack->buffers[i];
      30             :   }
      31             : 
      32          11 : }
      33             : 
      34             : 
      35        9375 : void netio_register_recv_buffer(struct netio_recv_socket* socket, struct netio_buffer* buf, uint64_t  flags)
      36             : {
      37        9375 :   if (socket->tcp_fi_mode != NETIO_MODE_LIBFABRIC) {
      38             :     return;
      39             :   }
      40             : 
      41             :   log_trc("registering recv buffer number: %d", socket->reg_mr);
      42             :   uint64_t access = FI_RECV;
      43             :   int ret = 0;
      44        9375 :   if((ret = fi_mr_reg(socket->domain, buf->data, buf->size, access, 0, socket->req_key++, 0, &buf->mr, NULL)))
      45             :   {
      46           0 :     log_fatal("Failed to register recv buffer (key %lu) failed. Error %d: %s",socket->req_key, ret, fi_strerror(-ret));
      47             :     //exit(2);
      48             :   }
      49             :   if(ret==0){
      50        9375 :     if (socket->reg_mr >= NETIO_DOMAIN_MAX_MR){
      51           0 :       log_fatal("Trying to register recv buffer number %u. MR array size %lu.", socket->reg_mr, NETIO_DOMAIN_MAX_MR);
      52           0 :       exit(1);
      53             :     }
      54        9375 :     socket->mr[socket->reg_mr]=&buf->mr->fid;
      55        9375 :     socket->reg_mr+=1;
      56             :   }
      57             : }
      58             : 
      59             : 
      60        1866 : void netio_register_send_buffer(struct netio_send_socket* socket, struct netio_buffer* buf, uint64_t flags)
      61             : {
      62        1866 :   if (socket->tcp_fi_mode != NETIO_MODE_LIBFABRIC) {
      63             :     return;
      64             :   }
      65             : 
      66             :   uint64_t access = FI_SEND;
      67             :   int ret = 0;
      68        1609 :   if((ret = fi_mr_reg(socket->domain->domain, buf->data, buf->size, access, 0, socket->domain->req_key++, 0, &buf->mr, NULL)))
      69             :   {
      70           0 :     log_fatal("Failed to register send buffer failed. Error %d: %s.", ret, fi_strerror(-ret));
      71           0 :     exit(2);
      72             :   }
      73             :   if(ret==0){
      74        1609 :     if (socket->domain->reg_mr >= NETIO_DOMAIN_MAX_MR){
      75           0 :       log_fatal("Trying to register send buffer number %u. MR array size %lu.", socket->domain->reg_mr, NETIO_DOMAIN_MAX_MR);
      76           0 :       exit(1);
      77             :     }
      78        1609 :     socket->domain->mr[socket->domain->reg_mr]=&buf->mr->fid;
      79        1609 :     socket->domain->reg_mr+=1;
      80             :   }
      81             : }
      82             : 
      83             : 
      84             : void
      85          11 : netio_bufferstack_send_init(struct netio_buffer_stack* stack, struct netio_send_socket* socket, size_t num, size_t bufsize, uint64_t flags)
      86             : {
      87             :   log_dbg("Initializing memory: %d pages", num);
      88          11 :   init_bufferstack(stack, num, bufsize);
      89          11 :   if (socket->tcp_fi_mode==NETIO_MODE_LIBFABRIC) {
      90           7 :     socket->domain->reg_mr=0;
      91        1311 :     for(unsigned i=0; i<num; i++) {
      92        1304 :       netio_register_send_buffer(socket, stack->buffers[i], flags);
      93             :     }
      94             :   }
      95          11 : }
      96             : 
      97             : void
      98           7 : netio_bufferstack_close(struct netio_buffer_stack* stack, size_t num)
      99             : {
     100             :   log_dbg("Freeing memory: %d pages", num);
     101        1551 :   for(unsigned int i=0; i < num; i++) {
     102        1544 :     if(stack->buffers == NULL){log_warn("Buffer already released."); return;}
     103        1544 :     free(stack->buffer_addresses[i]->data);
     104        1544 :     free(stack->buffer_addresses[i]);
     105             :   }
     106           7 :   free(stack->buffers);
     107           7 :   free(stack->buffer_addresses);
     108           7 :   stack->buffers = NULL;
     109             : }
     110             : 
     111             : 
     112             : int
     113          26 : netio_bufferstack_pop(struct netio_buffer_stack* stack, struct netio_buffer** buffer)
     114             : {
     115             :   log_trc("pop: available %lu", stack->available_buffers);
     116          26 :   if(stack->available_buffers > 0) {
     117          26 :     stack->available_buffers--;
     118          26 :     *buffer = stack->buffers[stack->available_buffers];
     119          26 :     return 0;
     120             :   } else {
     121             :     return 1;
     122             :   }
     123             : }
     124             : 
     125             : 
     126             : int
     127          26 : netio_bufferstack_push(struct netio_buffer_stack* stack, struct netio_buffer* buffer)
     128             : {
     129             :   log_trc("push: available %lu", stack->available_buffers);
     130          26 :   if(stack->available_buffers < stack->num_buffers) {
     131          26 :     stack->buffers[stack->available_buffers] = buffer;
     132          26 :     stack->available_buffers++;
     133          26 :     return 0;
     134             :   } else {
     135             :     return 1;
     136             :   }
     137             : }