pd.if.org Git - liblfds/blob - liblfds/liblfds7.1.0/test_and_benchmark/libtest/src/libtest_tests/libtest_tests_queue_unbounded_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_umm_state
  17     *qs;
  18 };
  19
  20 struct test_element
  21 {
  22   struct lfds710_queue_umm_element
  23     qe,
  24     *qe_use;
  25
  26   lfds710_pal_uint_t
  27     counter,
  28     thread_number;
  29 };
  30
  31 /***** private prototypes *****/
  32 static libshared_pal_thread_return_t LIBSHARED_PAL_THREAD_CALLING_CONVENTION thread_enqueuer_and_dequeuer( void *libtest_threadset_per_thread_state );
  33
  34
  35
  36
  37
  38 /****************************************************************************/
  39 void libtest_tests_queue_umm_enqueuing_and_dequeuing( struct lfds710_list_asu_state *list_of_logical_processors, struct libshared_memory_state *ms, enum lfds710_misc_validity *dvs )
  40 {
  41   lfds710_pal_uint_t
  42     loop,
  43     number_elements,
  44     number_logical_processors,
  45     *per_thread_counters,
  46     subloop;
  47
  48   struct lfds710_list_asu_element
  49     *lasue = NULL;
  50
  51   struct lfds710_queue_umm_state
  52     qs;
  53
  54   struct lfds710_misc_validation_info
  55     vi;
  56
  57   struct libtest_logical_processor
  58     *lp;
  59
  60   struct libtest_threadset_per_thread_state
  61     *pts;
  62
  63   struct libtest_threadset_state
  64     ts;
  65
  66   struct test_element
  67     *te_array;
  68
  69   struct test_per_thread_state
  70     *tpts;
  71
  72   LFDS710_PAL_ASSERT( list_of_logical_processors != NULL );
  73   LFDS710_PAL_ASSERT( ms != NULL );
  74   LFDS710_PAL_ASSERT( dvs != NULL );
  75
  76   /* TRD : create a queue with one element per thread
  77            each thread constly dequeues and enqueues from that one queue
  78            where when enqueuing sets in the element
  79            its thread number and counter
  80            and when dequeuing, checks the thread number and counter
  81            against previously seen counter for that thread
  82            where it should always see a higher number
  83   */
  84
  85   *dvs = LFDS710_MISC_VALIDITY_VALID;
  86
  87   // TRD : allocate
  88   lfds710_list_asu_query( list_of_logical_processors, LFDS710_LIST_ASU_QUERY_GET_POTENTIALLY_INACCURATE_COUNT, NULL, (void **) &number_logical_processors );
  89   tpts = libshared_memory_alloc_from_unknown_node( ms, sizeof(struct test_per_thread_state) * number_logical_processors, LFDS710_PAL_ATOMIC_ISOLATION_IN_BYTES );
  90   pts = libshared_memory_alloc_from_unknown_node( ms, sizeof(struct libtest_threadset_per_thread_state) * number_logical_processors, LFDS710_PAL_ATOMIC_ISOLATION_IN_BYTES );
  91   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 );
  92   te_array = libshared_memory_alloc_largest_possible_array_from_unknown_node( ms, sizeof(struct test_element), LFDS710_PAL_ATOMIC_ISOLATION_IN_BYTES, &number_elements );
  93
  94   lfds710_queue_umm_init_valid_on_current_logical_core( &qs, &(te_array+number_logical_processors)->qe, NULL );
  95
  96   // TRD : we assume the test will iterate at least once (or we'll have a false negative)
  97   for( loop = 0 ; loop < number_logical_processors ; loop++ )
  98   {
  99     (te_array+loop)->thread_number = loop;
 100     (te_array+loop)->counter = 0;
 101     LFDS710_QUEUE_UMM_SET_VALUE_IN_ELEMENT( (te_array+loop)->qe, te_array+loop );
 102     lfds710_queue_umm_enqueue( &qs, &(te_array+loop)->qe );
 103   }
 104
 105   libtest_threadset_init( &ts, NULL );
 106
 107   loop = 0;
 108
 109   while( LFDS710_LIST_ASU_GET_START_AND_THEN_NEXT(*list_of_logical_processors,lasue) )
 110   {
 111     lp = LFDS710_LIST_ASU_GET_VALUE_FROM_ELEMENT( *lasue );
 112     (tpts+loop)->qs = &qs;
 113     (tpts+loop)->thread_number = loop;
 114     (tpts+loop)->counter = 0;
 115     (tpts+loop)->error_flag = LOWERED;
 116     (tpts+loop)->per_thread_counters = per_thread_counters + loop * number_logical_processors;
 117     (tpts+loop)->number_logical_processors = number_logical_processors;
 118
 119     for( subloop = 0 ; subloop < number_logical_processors ; subloop++ )
 120       *((tpts+loop)->per_thread_counters+subloop) = 0;
 121
 122     libtest_threadset_add_thread( &ts, &pts[loop], lp, thread_enqueuer_and_dequeuer, &tpts[loop] );
 123     loop++;
 124   }
 125
 126   LFDS710_MISC_BARRIER_STORE;
 127
 128   lfds710_misc_force_store();
 129
 130   // TRD : run the test
 131   libtest_threadset_run( &ts );
 132
 133   libtest_threadset_cleanup( &ts );
 134
 135   // TRD : validate
 136   LFDS710_MISC_BARRIER_LOAD;
 137
 138   vi.min_elements = vi.max_elements = number_logical_processors;
 139
 140   lfds710_queue_umm_query( &qs, LFDS710_QUEUE_UMM_QUERY_SINGLETHREADED_VALIDATE, &vi, dvs );
 141
 142   for( loop = 0 ; loop < number_logical_processors ; loop++ )
 143     if( (tpts+loop)->error_flag == RAISED )
 144       *dvs = LFDS710_MISC_VALIDITY_INVALID_TEST_DATA;
 145
 146   lfds710_queue_umm_cleanup( &qs, NULL );
 147
 148   return;
 149 }
 150
 151
 152
 153
 154
 155 /****************************************************************************/
 156 static libshared_pal_thread_return_t LIBSHARED_PAL_THREAD_CALLING_CONVENTION thread_enqueuer_and_dequeuer( void *libtest_threadset_per_thread_state )
 157 {
 158   lfds710_pal_uint_t
 159     time_loop = 0;
 160
 161   struct lfds710_queue_umm_element
 162     *qe;
 163
 164   struct test_element
 165     *te;
 166
 167   struct test_per_thread_state
 168     *tpts;
 169
 170   struct libtest_threadset_per_thread_state
 171     *pts;
 172
 173   time_t
 174     current_time,
 175     start_time;
 176
 177   LFDS710_MISC_MAKE_VALID_ON_CURRENT_LOGICAL_CORE_INITS_COMPLETED_BEFORE_NOW_ON_ANY_OTHER_LOGICAL_CORE;
 178
 179   LFDS710_PAL_ASSERT( libtest_threadset_per_thread_state != NULL );
 180
 181   pts = (struct libtest_threadset_per_thread_state *) libtest_threadset_per_thread_state;
 182   tpts = LIBTEST_THREADSET_GET_USER_STATE_FROM_PER_THREAD_STATE( *pts );
 183
 184   libtest_threadset_thread_ready_and_wait( pts );
 185
 186   current_time = start_time = time( NULL );
 187
 188   while( current_time < start_time + TEST_DURATION_IN_SECONDS )
 189   {
 190     lfds710_queue_umm_dequeue( tpts->qs, &qe );
 191     te = LFDS710_QUEUE_UMM_GET_VALUE_FROM_ELEMENT( *qe );
 192
 193     if( te->thread_number >= tpts->number_logical_processors )
 194       tpts->error_flag = RAISED;
 195     else
 196     {
 197       if( te->counter < tpts->per_thread_counters[te->thread_number] )
 198         tpts->error_flag = RAISED;
 199
 200       if( te->counter >= tpts->per_thread_counters[te->thread_number] )
 201         tpts->per_thread_counters[te->thread_number] = te->counter+1;
 202     }
 203
 204     te->thread_number = tpts->thread_number;
 205     te->counter = ++tpts->counter;
 206
 207     LFDS710_QUEUE_UMM_SET_VALUE_IN_ELEMENT( *qe, te );
 208     lfds710_queue_umm_enqueue( tpts->qs, qe );
 209
 210     if( time_loop++ == TIME_LOOP_COUNT )
 211     {
 212       time_loop = 0;
 213       time( &current_time );
 214     }
 215   }
 216
 217   LFDS710_MISC_BARRIER_STORE;
 218
 219   lfds710_misc_force_store();
 220
 221   return LIBSHARED_PAL_THREAD_RETURN_CAST(RETURN_SUCCESS);
 222 }
 223