caveat/shmfifo.h

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