trunk/verif/md/model_03.spin

   1 /* LTTng ltt-tracer.c atomic lockless buffering scheme Promela model v2
   2  * Created for the Spin validator.
   3  * Mathieu Desnoyers <mathieu.desnoyers@polymtl.ca>
   4  * October 2008
   5
   6  * TODO : create test cases that will generate an overflow on the offset and
   7  * counter type. Counter types smaller than a byte should be used.
   8
   9  * Promela only has unsigned char, no signed char.
  10  * Because detection of difference < 0 depends on a signed type, but we want
  11  * compactness, check also for the values being higher than half of the unsigned
  12  * char range (and consider them negative). The model, by design, does not use
  13  * offsets or counts higher than 127 because we would then have to use a larger
  14  * type (short or int).
  15  */
  16 #define HALF_UCHAR (255/2)
  17
  18 /* NUMPROCS 4 : causes event loss with some reader timings.
  19  * e.g. 3 events, 1 switch, 1 event (lost, buffer full), read 1 subbuffer
  20  */
  21 #define NUMPROCS 4
  22
  23 /* NUMPROCS 3 : does not cause event loss because buffers are big enough.
  24  * #define NUMPROCS 3
  25  * e.g. 3 events, 1 switch, read 1 subbuffer
  26  */
  27
  28 #define NUMSWITCH 1
  29 #define BUFSIZE 4
  30 #define NR_SUBBUFS 2
  31 #define SUBBUF_SIZE (BUFSIZE / NR_SUBBUFS)
  32
  33 /* Writer counters
  34 */
  35 byte write_off = 0;
  36 byte commit_count[NR_SUBBUFS];
  37
  38 /* <formal_verif> */
  39 byte _commit_sum;
  40 /* </formal_verif> */
  41
  42 /* Reader counters
  43 */
  44 byte read_off = 0;
  45
  46 byte events_lost = 0;
  47 byte refcount = 0;
  48
  49 bool deliver = 0;
  50
  51 //#ifdef RACE_TEST
  52 /* buffer slot in-use bit. Detects racy use (more than a single process
  53  * accessing a slot at any given step).
  54  */
  55 bool buffer_use[BUFSIZE];
  56 //#endif
  57
  58 /* Proceed to a sub-subber switch is needed.
  59  * Used in a periodical timer interrupt to fill and ship the current subbuffer
  60  * to the reader so we can guarantee a steady flow. If a subbuffer is
  61  * completely empty, do not switch.
  62  * Also used as "finalize" operation to complete the last subbuffer after
  63  * all writers have finished so the last subbuffer can be read by the reader.
  64  */
  65 proctype switcher()
  66 {
  67   byte prev_off, new_off, tmp_commit;
  68   byte size;
  69
  70 cmpxchg_loop:
  71   atomic {
  72     prev_off = write_off;
  73     size = SUBBUF_SIZE - (prev_off % SUBBUF_SIZE);
  74     new_off = prev_off + size;
  75     if
  76       :: (new_off - read_off > BUFSIZE && new_off - read_off < HALF_UCHAR)
  77          || size == SUBBUF_SIZE ->
  78          refcount = refcount - 1;
  79          goto not_needed;
  80       :: else -> skip
  81     fi;
  82   }
  83   atomic {
  84     if
  85       :: prev_off != write_off -> goto cmpxchg_loop
  86       :: else -> write_off = new_off;
  87     fi;
  88   }
  89
  90   atomic {
  91     tmp_commit = commit_count[(prev_off % BUFSIZE) / SUBBUF_SIZE] + size;
  92     /* <formal_verif> */
  93     _commit_sum = _commit_sum - commit_count[(prev_off % BUFSIZE) / SUBBUF_SIZE]
  94                   + tmp_commit;
  95     /* </formal_verif> */
  96     commit_count[(prev_off % BUFSIZE) / SUBBUF_SIZE] = tmp_commit;
  97     if
  98       :: (((prev_off / BUFSIZE) * BUFSIZE) / NR_SUBBUFS) + SUBBUF_SIZE -
  99          tmp_commit
 100          -> deliver = 1
 101       :: else
 102          -> skip
 103     fi;
 104     refcount = refcount - 1;
 105   }
 106 not_needed:
 107   skip;
 108 }
 109
 110 /* tracer
 111  * Writes 1 byte of information in the buffer at the current
 112  * "write_off" position and then increment the commit_count of the sub-buffer
 113  * the information has been written to.
 114  */
 115 proctype tracer()
 116 {
 117   byte size = 1;
 118   byte prev_off, new_off, tmp_commit;
 119   byte i, j;
 120
 121 cmpxchg_loop:
 122   atomic {
 123     prev_off = write_off;
 124     new_off = prev_off + size;
 125   }
 126   atomic {
 127     if
 128       :: new_off - read_off > BUFSIZE && new_off - read_off < HALF_UCHAR ->
 129          goto lost
 130       :: else -> skip
 131     fi;
 132   }
 133   atomic {
 134     if
 135       :: prev_off != write_off -> goto cmpxchg_loop
 136       :: else -> write_off = new_off;
 137     fi;
 138     i = 0;
 139     do
 140       :: i < size ->
 141          assert(buffer_use[(prev_off + i) % BUFSIZE] == 0);
 142          buffer_use[(prev_off + i) % BUFSIZE] = 1;
 143          i++
 144       :: i >= size -> break
 145     od;
 146   }
 147
 148   /* writing to buffer...
 149   */
 150
 151   atomic {
 152     i = 0;
 153     do
 154       :: i < size ->
 155          buffer_use[(prev_off + i) % BUFSIZE] = 0;
 156          i++
 157       :: i >= size -> break
 158     od;
 159     tmp_commit = commit_count[(prev_off % BUFSIZE) / SUBBUF_SIZE] + size;
 160     /* <formal_verif> */
 161     _commit_sum = _commit_sum - commit_count[(prev_off % BUFSIZE) / SUBBUF_SIZE]
 162                   + tmp_commit;
 163     /* </formal_verif> */
 164     commit_count[(prev_off % BUFSIZE) / SUBBUF_SIZE] = tmp_commit;
 165     if
 166       :: (((prev_off / BUFSIZE) * BUFSIZE) / NR_SUBBUFS) + SUBBUF_SIZE -
 167          tmp_commit
 168          -> deliver = 1
 169       :: else
 170          -> skip
 171     fi;
 172   }
 173   atomic {
 174     goto end;
 175 lost:
 176     events_lost++;
 177 end:
 178     refcount = refcount - 1;
 179   }
 180 }
 181
 182 /* reader
 183  * Read the information sub-buffer per sub-buffer when available.
 184  *
 185  * Reads the information as soon as it is ready, or may be delayed by
 186  * an asynchronous delivery. Being modeled as a process insures all cases
 187  * (scheduled very quickly or very late, causing event loss) are covered.
 188  * Only one reader per buffer (normally ensured by a mutex). This is modeled
 189  * by using a single reader process.
 190  */
 191 proctype reader()
 192 {
 193   byte i, j;
 194
 195   do
 196     :: (write_off / SUBBUF_SIZE) - (read_off / SUBBUF_SIZE) > 0
 197        && (write_off / SUBBUF_SIZE) - (read_off / SUBBUF_SIZE) < HALF_UCHAR
 198        && (commit_count[(read_off % BUFSIZE) / SUBBUF_SIZE]
 199            - SUBBUF_SIZE - (((read_off / BUFSIZE) * BUFSIZE) / NR_SUBBUFS)
 200            == 0) ->
 201        atomic {
 202          i = 0;
 203          do
 204            :: i < SUBBUF_SIZE ->
 205               assert(buffer_use[(read_off + i) % BUFSIZE] == 0);
 206               buffer_use[(read_off + i) % BUFSIZE] = 1;
 207               i++
 208            :: i >= SUBBUF_SIZE -> break
 209          od;
 210        }
 211
 212        /* reading from buffer...
 213        */
 214
 215        atomic {
 216          i = 0;
 217          do
 218            :: i < SUBBUF_SIZE ->
 219               buffer_use[(read_off + i) % BUFSIZE] = 0;
 220               i++
 221            :: i >= SUBBUF_SIZE -> break
 222          od;
 223          read_off = read_off + SUBBUF_SIZE;
 224        }
 225     :: read_off >= (NUMPROCS - events_lost) -> break;
 226   od;
 227 }
 228
 229 /* Waits for all tracer and switcher processes to finish before finalizing
 230  * the buffer. Only after that will the reader be allowed to read the
 231  * last subbuffer.
 232  */
 233 proctype cleaner()
 234 {
 235   atomic {
 236     do
 237       :: refcount == 0 ->
 238          refcount = refcount + 1;
 239          run switcher();  /* Finalize the last sub-buffer so it can be read. */
 240          break;
 241     od;
 242   }
 243 }
 244
 245 init {
 246   byte i = 0;
 247   byte j = 0;
 248   byte sum = 0;
 249   byte commit_sum = 0;
 250
 251   atomic {
 252     i = 0;
 253     do
 254       :: i < NR_SUBBUFS ->
 255          commit_count[i] = 0;
 256          i++
 257       :: i >= NR_SUBBUFS -> break
 258     od;
 259     /* <formal_verif> */
 260     _commit_sum = 0;
 261     /* </formal_verif> */
 262     i = 0;
 263     do
 264       :: i < BUFSIZE ->
 265          buffer_use[i] = 0;
 266          i++
 267       :: i >= BUFSIZE -> break
 268     od;
 269     run reader();
 270     run cleaner();
 271     i = 0;
 272     do
 273       :: i < NUMPROCS ->
 274          refcount = refcount + 1;
 275          run tracer();
 276          i++
 277       :: i >= NUMPROCS -> break
 278     od;
 279     i = 0;
 280     do
 281       :: i < NUMSWITCH ->
 282          refcount = refcount + 1;
 283          run switcher();
 284          i++
 285       :: i >= NUMSWITCH -> break
 286     od;
 287   }
 288 }
 289