pd.if.org Git - liblfds/blob - liblfds/liblfds6.1.0/liblfds610/src/lfds610_queue/lfds610_queue_queue.c

   1 #include "lfds610_queue_internal.h"
   2
   3
   4
   5
   6
   7 /****************************************************************************/
   8 int lfds610_queue_enqueue( struct lfds610_queue_state *qs, void *user_data )
   9 {
  10   LFDS610_ALIGN(LFDS610_ALIGN_DOUBLE_POINTER) struct lfds610_queue_element
  11     *qe[LFDS610_QUEUE_PAC_SIZE];
  12
  13   assert( qs != NULL );
  14   // TRD : user_data can be NULL
  15
  16   lfds610_queue_internal_new_element_from_freelist( qs, qe, user_data );
  17
  18   if( qe[LFDS610_QUEUE_POINTER] == NULL )
  19     return( 0 );
  20
  21   lfds610_queue_internal_queue( qs, qe );
  22
  23   return( 1 );
  24 }
  25
  26
  27
  28
  29
  30 /****************************************************************************/
  31 int lfds610_queue_guaranteed_enqueue( struct lfds610_queue_state *qs, void *user_data )
  32 {
  33   LFDS610_ALIGN(LFDS610_ALIGN_DOUBLE_POINTER) struct lfds610_queue_element
  34     *qe[LFDS610_QUEUE_PAC_SIZE];
  35
  36   assert( qs != NULL );
  37   // TRD : user_data can be NULL
  38
  39   lfds610_queue_internal_guaranteed_new_element_from_freelist( qs, qe, user_data );
  40
  41   if( qe[LFDS610_QUEUE_POINTER] == NULL )
  42     return( 0 );
  43
  44   lfds610_queue_internal_queue( qs, qe );
  45
  46   return( 1 );
  47 }
  48
  49
  50
  51
  52
  53 /****************************************************************************/
  54 void lfds610_queue_internal_queue( struct lfds610_queue_state *qs, struct lfds610_queue_element *qe[LFDS610_QUEUE_PAC_SIZE] )
  55 {
  56   LFDS610_ALIGN(LFDS610_ALIGN_DOUBLE_POINTER) struct lfds610_queue_element
  57     *enqueue[LFDS610_QUEUE_PAC_SIZE],
  58     *next[LFDS610_QUEUE_PAC_SIZE];
  59
  60   unsigned char
  61     cas_result = 0;
  62
  63   assert( qs != NULL );
  64   assert( qe != NULL );
  65
  66   // TRD : the DCAS operation issues a read and write barrier, so we don't need a read barrier in the do() loop
  67
  68   LFDS610_BARRIER_LOAD;
  69
  70   do
  71   {
  72     enqueue[LFDS610_QUEUE_POINTER] = qs->enqueue[LFDS610_QUEUE_POINTER];
  73     enqueue[LFDS610_QUEUE_COUNTER] = qs->enqueue[LFDS610_QUEUE_COUNTER];
  74
  75     next[LFDS610_QUEUE_POINTER] = enqueue[LFDS610_QUEUE_POINTER]->next[LFDS610_QUEUE_POINTER];
  76     next[LFDS610_QUEUE_COUNTER] = enqueue[LFDS610_QUEUE_POINTER]->next[LFDS610_QUEUE_COUNTER];
  77
  78     /* TRD : this if() ensures that the next we read, just above,
  79              really is from qs->enqueue (which we copied into enqueue)
  80     */
  81
  82     LFDS610_BARRIER_LOAD;
  83
  84     if( qs->enqueue[LFDS610_QUEUE_POINTER] == enqueue[LFDS610_QUEUE_POINTER] and qs->enqueue[LFDS610_QUEUE_COUNTER] == enqueue[LFDS610_QUEUE_COUNTER] )
  85     {
  86       if( next[LFDS610_QUEUE_POINTER] == NULL )
  87       {
  88         qe[LFDS610_QUEUE_COUNTER] = next[LFDS610_QUEUE_COUNTER] + 1;
  89         cas_result = lfds610_abstraction_dcas( (volatile lfds610_atom_t *) enqueue[LFDS610_QUEUE_POINTER]->next, (lfds610_atom_t *) qe, (lfds610_atom_t *) next );
  90       }
  91       else
  92       {
  93         next[LFDS610_QUEUE_COUNTER] = enqueue[LFDS610_QUEUE_COUNTER] + 1;
  94         lfds610_abstraction_dcas( (volatile lfds610_atom_t *) qs->enqueue, (lfds610_atom_t *) next, (lfds610_atom_t *) enqueue );
  95       }
  96     }
  97   }
  98   while( cas_result == 0 );
  99
 100   qe[LFDS610_QUEUE_COUNTER] = enqueue[LFDS610_QUEUE_COUNTER] + 1;
 101   lfds610_abstraction_dcas( (volatile lfds610_atom_t *) qs->enqueue, (lfds610_atom_t *) qe, (lfds610_atom_t *) enqueue );
 102
 103   return;
 104 }
 105
 106
 107
 108
 109
 110 /****************************************************************************/
 111 int lfds610_queue_dequeue( struct lfds610_queue_state *qs, void **user_data )
 112 {
 113   LFDS610_ALIGN(LFDS610_ALIGN_DOUBLE_POINTER) struct lfds610_queue_element
 114     *enqueue[LFDS610_QUEUE_PAC_SIZE],
 115     *dequeue[LFDS610_QUEUE_PAC_SIZE],
 116     *next[LFDS610_QUEUE_PAC_SIZE];
 117
 118   unsigned char
 119     cas_result = 0;
 120
 121   int
 122     rv = 1,
 123     state = LFDS610_QUEUE_STATE_UNKNOWN,
 124     finished_flag = LOWERED;
 125
 126   assert( qs != NULL );
 127   assert( user_data != NULL );
 128
 129   // TRD : the DCAS operation issues a read and write barrier, so we don't need a read barrier in the do() loop
 130
 131   LFDS610_BARRIER_LOAD;
 132
 133   do
 134   {
 135     dequeue[LFDS610_QUEUE_POINTER] = qs->dequeue[LFDS610_QUEUE_POINTER];
 136     dequeue[LFDS610_QUEUE_COUNTER] = qs->dequeue[LFDS610_QUEUE_COUNTER];
 137
 138     enqueue[LFDS610_QUEUE_POINTER] = qs->enqueue[LFDS610_QUEUE_POINTER];
 139     enqueue[LFDS610_QUEUE_COUNTER] = qs->enqueue[LFDS610_QUEUE_COUNTER];
 140
 141     next[LFDS610_QUEUE_POINTER] = dequeue[LFDS610_QUEUE_POINTER]->next[LFDS610_QUEUE_POINTER];
 142     next[LFDS610_QUEUE_COUNTER] = dequeue[LFDS610_QUEUE_POINTER]->next[LFDS610_QUEUE_COUNTER];
 143
 144     /* TRD : confirm that dequeue didn't move between reading it
 145              and reading its next pointer
 146     */
 147
 148     LFDS610_BARRIER_LOAD;
 149
 150     if( dequeue[LFDS610_QUEUE_POINTER] == qs->dequeue[LFDS610_QUEUE_POINTER] and dequeue[LFDS610_QUEUE_COUNTER] == qs->dequeue[LFDS610_QUEUE_COUNTER] )
 151     {
 152       if( enqueue[LFDS610_QUEUE_POINTER] == dequeue[LFDS610_QUEUE_POINTER] and next[LFDS610_QUEUE_POINTER] == NULL )
 153         state = LFDS610_QUEUE_STATE_EMPTY;
 154
 155       if( enqueue[LFDS610_QUEUE_POINTER] == dequeue[LFDS610_QUEUE_POINTER] and next[LFDS610_QUEUE_POINTER] != NULL )
 156         state = LFDS610_QUEUE_STATE_ENQUEUE_OUT_OF_PLACE;
 157
 158       if( enqueue[LFDS610_QUEUE_POINTER] != dequeue[LFDS610_QUEUE_POINTER] )
 159         state = LFDS610_QUEUE_STATE_ATTEMPT_DELFDS610_QUEUE;
 160
 161       switch( state )
 162       {
 163         case LFDS610_QUEUE_STATE_EMPTY:
 164           *user_data = NULL;
 165           rv = 0;
 166           finished_flag = RAISED;
 167         break;
 168
 169         case LFDS610_QUEUE_STATE_ENQUEUE_OUT_OF_PLACE:
 170           next[LFDS610_QUEUE_COUNTER] = enqueue[LFDS610_QUEUE_COUNTER] + 1;
 171           lfds610_abstraction_dcas( (volatile lfds610_atom_t *) qs->enqueue, (lfds610_atom_t *) next, (lfds610_atom_t *) enqueue );
 172         break;
 173
 174         case LFDS610_QUEUE_STATE_ATTEMPT_DELFDS610_QUEUE:
 175           *user_data = next[LFDS610_QUEUE_POINTER]->user_data;
 176
 177           next[LFDS610_QUEUE_COUNTER] = dequeue[LFDS610_QUEUE_COUNTER] + 1;
 178           cas_result = lfds610_abstraction_dcas( (volatile lfds610_atom_t *) qs->dequeue, (lfds610_atom_t *) next, (lfds610_atom_t *) dequeue );
 179
 180           if( cas_result == 1 )
 181             finished_flag = RAISED;
 182         break;
 183       }
 184     }
 185   }
 186   while( finished_flag == LOWERED );
 187
 188   if( cas_result == 1 )
 189     lfds610_freelist_push( qs->fs, dequeue[LFDS610_QUEUE_POINTER]->fe );
 190
 191   return( rv );
 192 }
 193