pd.if.org Git - liblfds/blob - liblfds/liblfds7.1.0/test_and_benchmark/libtest/src/libtest_tests/libtest_tests_queue_bounded_manyproducer_manyconsumer_enqueuing_and_dequeuing.c

   1 /***** includes *****/
   2 #include "libtest_tests_internal.h"
   3
   4 /***** structs *****/
   5 struct test_per_thread_state
   6 {
   7   enum flag
   8     error_flag;
   9
  10   lfds710_pal_uint_t
  11     counter,
  12     number_logical_processors,
  13     *per_thread_counters,
  14     thread_number;
  15
  16   struct lfds710_queue_bmm_state
  17     *qbmms;
  18 };
  19
  20 /***** private prototypes *****/
  21 static libshared_pal_thread_return_t LIBSHARED_PAL_THREAD_CALLING_CONVENTION thread_enqueuer_and_dequeuer( void *libtest_threadset_per_thread_state );
  22
  23
  24
  25
  26
  27 /****************************************************************************/
  28 void libtest_tests_queue_bmm_enqueuing_and_dequeuing( struct lfds710_list_asu_state *list_of_logical_processors, struct libshared_memory_state *ms, enum lfds710_misc_validity *dvs )
  29 {
  30   lfds710_pal_uint_t
  31     loop,
  32     number_elements,
  33     number_logical_processors,
  34     *per_thread_counters,
  35     power_of_two_number_elements = 1,
  36     subloop;
  37
  38   struct lfds710_list_asu_element
  39     *lasue = NULL;
  40
  41   struct lfds710_queue_bmm_element
  42     *qbmme_array;
  43
  44   struct lfds710_queue_bmm_state
  45     qbmms;
  46
  47   struct lfds710_misc_validation_info
  48     vi;
  49
  50   struct libtest_logical_processor
  51     *lp;
  52
  53   struct libtest_threadset_per_thread_state
  54     *pts;
  55
  56   struct libtest_threadset_state
  57     ts;
  58
  59   struct test_per_thread_state
  60     *tpts;
  61
  62   LFDS710_PAL_ASSERT( list_of_logical_processors != NULL );
  63   LFDS710_PAL_ASSERT( ms != NULL );
  64   LFDS710_PAL_ASSERT( dvs != NULL );
  65
  66   /* TRD : create a queue with one element per thread
  67            each thread constly dequeues and enqueues from that one queue
  68            where when enqueuing sets in the element
  69            its thread number and counter
  70            and when dequeuing, checks the thread number and counter
  71            against previously seen counter for that thread
  72            where it should always see a higher number
  73   */
  74
  75   *dvs = LFDS710_MISC_VALIDITY_VALID;
  76
  77   // TRD : allocate
  78   lfds710_list_asu_query( list_of_logical_processors, LFDS710_LIST_ASU_QUERY_GET_POTENTIALLY_INACCURATE_COUNT, NULL, (void **) &number_logical_processors );
  79   tpts = libshared_memory_alloc_from_unknown_node( ms, sizeof(struct test_per_thread_state) * number_logical_processors, LFDS710_PAL_ATOMIC_ISOLATION_IN_BYTES );
  80   pts = libshared_memory_alloc_from_unknown_node( ms, sizeof(struct libtest_threadset_per_thread_state) * number_logical_processors, LFDS710_PAL_ATOMIC_ISOLATION_IN_BYTES );
  81   per_thread_counters = libshared_memory_alloc_from_unknown_node( ms, sizeof(lfds710_pal_uint_t) * number_logical_processors * number_logical_processors, LFDS710_PAL_ATOMIC_ISOLATION_IN_BYTES );
  82   qbmme_array = libshared_memory_alloc_largest_possible_array_from_unknown_node( ms, sizeof(struct lfds710_queue_bmm_element), LFDS710_PAL_ATOMIC_ISOLATION_IN_BYTES, &number_elements );
  83
  84   // TRD : need to only use a power of 2 number of elements
  85   number_elements >>= 1;
  86
  87   while( number_elements != 0 )
  88   {
  89     number_elements >>= 1;
  90     power_of_two_number_elements <<= 1;
  91   }
  92
  93   /* TRD : to make the test more demanding, smallest number of elements (greater than number of logical cores)
  94            we really want one element per core
  95            but with the power-of-2 requirements, we can't have it
  96   */
  97
  98   while( power_of_two_number_elements > number_logical_processors )
  99     power_of_two_number_elements >>= 1;
 100
 101   lfds710_queue_bmm_init_valid_on_current_logical_core( &qbmms, qbmme_array, power_of_two_number_elements, NULL );
 102
 103   // TRD : we assume the test will iterate at least once (or we'll have a false negative)
 104   for( loop = 0 ; loop < number_logical_processors ; loop++ )
 105     lfds710_queue_bmm_enqueue( &qbmms, (void *) loop, (void *) 0 );
 106
 107   libtest_threadset_init( &ts, NULL );
 108
 109   loop = 0;
 110
 111   while( LFDS710_LIST_ASU_GET_START_AND_THEN_NEXT(*list_of_logical_processors,lasue) )
 112   {
 113     lp = LFDS710_LIST_ASU_GET_VALUE_FROM_ELEMENT( *lasue );
 114     (tpts+loop)->qbmms = &qbmms;
 115     (tpts+loop)->thread_number = loop;
 116     (tpts+loop)->counter = 0;
 117     (tpts+loop)->error_flag = LOWERED;
 118     (tpts+loop)->per_thread_counters = per_thread_counters + loop * number_logical_processors;
 119     (tpts+loop)->number_logical_processors = number_logical_processors;
 120
 121     for( subloop = 0 ; subloop < number_logical_processors ; subloop++ )
 122       *((tpts+loop)->per_thread_counters+subloop) = 0;
 123
 124     libtest_threadset_add_thread( &ts, &pts[loop], lp, thread_enqueuer_and_dequeuer, &tpts[loop] );
 125     loop++;
 126   }
 127
 128   LFDS710_MISC_BARRIER_STORE;
 129
 130   lfds710_misc_force_store();
 131
 132   // TRD : run the test
 133   libtest_threadset_run( &ts );
 134
 135   libtest_threadset_cleanup( &ts );
 136
 137   // TRD : validate
 138   LFDS710_MISC_BARRIER_LOAD;
 139
 140   vi.min_elements = vi.max_elements = power_of_two_number_elements;
 141
 142   lfds710_queue_bmm_query( &qbmms, LFDS710_QUEUE_BMM_QUERY_SINGLETHREADED_VALIDATE, &vi, dvs );
 143
 144   for( loop = 0 ; loop < number_logical_processors ; loop++ )
 145     if( (tpts+loop)->error_flag == RAISED )
 146       *dvs = LFDS710_MISC_VALIDITY_INVALID_TEST_DATA;
 147
 148   lfds710_queue_bmm_cleanup( &qbmms, NULL );
 149
 150   return;
 151 }
 152
 153
 154
 155
 156
 157 /****************************************************************************/
 158 static libshared_pal_thread_return_t LIBSHARED_PAL_THREAD_CALLING_CONVENTION thread_enqueuer_and_dequeuer( void *libtest_threadset_per_thread_state )
 159 {
 160   lfds710_pal_uint_t
 161     counter,
 162     thread_number,
 163     time_loop = 0;
 164
 165   struct test_per_thread_state
 166     *tpts;
 167
 168   struct libtest_threadset_per_thread_state
 169     *pts;
 170
 171   time_t
 172     current_time,
 173     start_time;
 174
 175   LFDS710_MISC_MAKE_VALID_ON_CURRENT_LOGICAL_CORE_INITS_COMPLETED_BEFORE_NOW_ON_ANY_OTHER_LOGICAL_CORE;
 176
 177   LFDS710_PAL_ASSERT( libtest_threadset_per_thread_state != NULL );
 178
 179   pts = (struct libtest_threadset_per_thread_state *) libtest_threadset_per_thread_state;
 180   tpts = LIBTEST_THREADSET_GET_USER_STATE_FROM_PER_THREAD_STATE( *pts );
 181
 182   libtest_threadset_thread_ready_and_wait( pts );
 183
 184   current_time = start_time = time( NULL );
 185
 186   while( current_time < start_time + TEST_DURATION_IN_SECONDS )
 187   {
 188     /* TRD : this is a soft queue, so dequeue/enqueue operations although always occurring
 189              may not be visible by the time the enqueue/dequeue function returns
 190              i.e. all threads may have dequeued, and then enqueued, but not seen each others enqueues yet
 191              so the queue looks empty
 192     */
 193
 194     while( 0 == lfds710_queue_bmm_dequeue(tpts->qbmms, (void *) &thread_number, (void *) &counter) );
 195
 196     if( thread_number >= tpts->number_logical_processors )
 197       tpts->error_flag = RAISED;
 198     else
 199     {
 200       if( counter < tpts->per_thread_counters[thread_number] )
 201         tpts->error_flag = RAISED;
 202
 203       if( counter >= tpts->per_thread_counters[thread_number] )
 204         tpts->per_thread_counters[thread_number] = counter+1;
 205     }
 206
 207     thread_number = tpts->thread_number;
 208     counter = ++tpts->counter;
 209
 210     // TRD : the enqueue can only succeed once a dequeue of the *very next element* in the queue has become visible (i.e. our down earlier dequeue may not be the right dequeue)
 211     while( 0 == lfds710_queue_bmm_enqueue(tpts->qbmms, (void *) thread_number, (void *) counter) );
 212
 213     if( time_loop++ == TIME_LOOP_COUNT )
 214     {
 215       time_loop = 0;
 216       time( &current_time );
 217     }
 218   }
 219
 220   LFDS710_MISC_BARRIER_STORE;
 221
 222   lfds710_misc_force_store();
 223
 224   return LIBSHARED_PAL_THREAD_RETURN_CAST(RETURN_SUCCESS);
 225 }
 226