caveat/shmfifo.h

   1 /*
   2   Copyright (c) 2020 Peter Hsu.  All Rights Reserved.  See LICENCE file for details.
   3 */
   4
   5 #include <unistd.h>
   6 #include <sys/syscall.h>
   7 #include <linux/futex.h>
   8 #include <time.h>
   9 #include <immintrin.h>
  10
  11
  12 #define BATCH_SIZE  256
  13 #define MAX_SPINS  100
  14
  15 struct fifo_t {
  16   const char* id;               /* descriptor, $name or path */
  17   int32_t head;                 /* removal pointer (local copy) */
  18   volatile int32_t TAIL;        /* global copy of insertion pointer */
  19   uint32_t get_mask;            /* =(1<<size)-1 */
  20   int32_t size;                 /* log-base-2 number of elements */
  21   int32_t pad1[32-6];           /* to 64-byte cache line */
  22   int32_t tail;                 /* insertion pointer (local copy) */
  23   volatile int32_t HEAD;        /* global copy of removal pointer */
  24   uint32_t put_mask;            /* another copy in producer cache line */
  25   int32_t fd;                   /* file descriptor number */
  26   volatile int32_t finished;    /* set flag when consumer is done */
  27   int32_t pad2[32-5];           /* to 64-byte cache line */
  28   volatile uint64_t buffer[0];  /* begining of buffer */
  29 };
  30
  31 static struct timespec timeout = { .tv_sec=0, .tv_nsec=100 };
  32
  33 //#define futex_wait(addr, val) syscall(SYS_futex, addr, FUTEX_WAIT, val, &timeout, 0)
  34 #define futex_wait(addr, val) syscall(SYS_futex, addr, FUTEX_WAIT, val,  (int*)0)
  35 #define futex_hibernate(addr, val) syscall(SYS_futex, addr, FUTEX_WAIT, val, (int*)0)
  36 #define futex_wake(addr) syscall(SYS_futex, addr, FUTEX_WAKE, 1)
  37
  38
  39 struct fifo_t* fifo_create( const char* bufid, int bufsize );
  40 /* Producer side fifo initialization.
  41      bufid      - number = file descriptor (already opened)
  42                   $name = shared memory segment /dev/shm/name
  43                   otherwise = trace file path name
  44      bufsize    - log-base-2 number of bytes
  45 */
  46 struct fifo_t* fifo_open( const char* bufid );
  47 /* Consumer side fifo initialization.
  48      bufid      - number = file descriptor (already opened)
  49                   $name = shared memory segment /dev/shm/name
  50                   otherwise = trace file path name
  51 */
  52
  53 void fifo_finish( struct fifo_t* fifo );
  54 /* Producer side fifo termination. */
  55 void fifo_close( struct fifo_t* fifo );
  56 /* Consumer side fifo termination. */
  57
  58
  59 void fifo_debug( struct fifo_t* fifo, const char* msg );
  60
  61
  62 /* Put item in fifo */
  63 static inline void fifo_put( struct fifo_t* fifo, uint64_t item )
  64 {
  65   int tailp1 = (fifo->tail+1) & fifo->put_mask;
  66   if (tailp1 == fifo->HEAD) {
  67     int spins = MAX_SPINS;
  68     do {
  69       _mm_pause();
  70     } while (tailp1 == fifo->HEAD && --spins >= 0);
  71     while (tailp1 == fifo->HEAD)
  72       futex_wait(&fifo->HEAD, tailp1);
  73   }
  74   fifo->buffer[fifo->tail] = item;
  75   fifo->tail = tailp1;
  76   if (fifo->tail % BATCH_SIZE == 0) {
  77     fifo->TAIL = fifo->tail;
  78     futex_wake(&fifo->TAIL);
  79   }
  80 }
  81
  82
  83 /* Get item from fifo */
  84 static inline uint64_t fifo_get( struct fifo_t* fifo )
  85 {
  86   if (fifo->head == fifo->TAIL) {
  87     int spins = MAX_SPINS;
  88     do {
  89       _mm_pause();
  90     } while (fifo->head == fifo->TAIL && --spins >= 0);
  91     while (fifo->head == fifo->TAIL)
  92       futex_wait(&fifo->TAIL, fifo->head);
  93   }
  94   uint64_t rv = fifo->buffer[fifo->head++];
  95   fifo->head &= fifo->get_mask;
  96   if (fifo->head % BATCH_SIZE == 0) {
  97     fifo->HEAD = fifo->head;
  98     futex_wake(&fifo->HEAD);
  99   }
 100   return rv;
 101 }
 102
 103
 104
 105
 106
 107 /* Make consumer status up to date */
 108 static inline void fifo_flush( struct fifo_t* fifo )
 109 {
 110   fifo->TAIL = fifo->tail;
 111   futex_wake(&fifo->TAIL);
 112 }