← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to mysys/thr_lock.cc

  • Committer: Monty Taylor
  • Date: 2009-09-30 07:01:32 UTC
  • mto: This revision was merged to the branch mainline in revision 1184.
  • Revision ID: mordred@inaugust.com-20090930070132-b1ol1xu1rpajdddy
Small namespace cleanup.

Show diffs side-by-side

added added

removed removed

Lines of Context:
mysys/thr_lock.cc
26
26
 
27
27
TL_READ                 # Low priority read
28
28
TL_READ_WITH_SHARED_LOCKS
29
 
TL_READ_HIGH_PRIORITY   # High priority read
30
29
TL_READ_NO_INSERT       # Read without concurrent inserts
31
30
TL_WRITE_ALLOW_WRITE    # Write lock that allows other writers
32
31
TL_WRITE_ALLOW_READ     # Write lock, but allow reading
33
32
TL_WRITE_CONCURRENT_INSERT
34
33
                        # Insert that can be mixed when selects
35
 
TL_WRITE_DELAYED        # Used by delayed insert
36
 
                        # Allows lower locks to take over
37
 
TL_WRITE_LOW_PRIORITY   # Low priority write
38
34
TL_WRITE                # High priority write
39
35
TL_WRITE_ONLY           # High priority write
40
36
                        # Abort all new lock request with an error
65
61
        for concurrent reads.
66
62
 
67
63
The lock algorithm allows one to have one TL_WRITE_ALLOW_READ,
68
 
TL_WRITE_CONCURRENT_INSERT or one TL_WRITE_DELAYED lock at the same time as
69
 
multiple read locks.
 
64
TL_WRITE_CONCURRENT_INSERT lock at the same time as multiple read locks.
70
65
 
71
66
*/
72
67
 
73
 
#include "mysys_priv.h"
 
68
#include "mysys/mysys_priv.h"
74
69
 
75
70
#include "thr_lock.h"
76
71
#include <mystrings/m_string.h>
77
72
#include <errno.h>
 
73
#include <list>
78
74
 
79
75
#if TIME_WITH_SYS_TIME
80
76
# include <sys/time.h>
85
81
# else
86
82
#  include <time.h>
87
83
# endif
88
 
#endif  
 
84
#endif
89
85
 
90
86
#include <drizzled/util/test.h>
91
87
 
92
 
bool thr_lock_inited=0;
 
88
using namespace std;
 
89
 
 
90
bool thr_lock_inited= false;
93
91
uint32_t locks_immediate = 0L, locks_waited = 0L;
94
 
ulong table_lock_wait_timeout;
 
92
uint64_t table_lock_wait_timeout;
95
93
enum thr_lock_type thr_upgraded_concurrent_insert_lock = TL_WRITE;
96
94
 
97
95
 
98
 
LIST *thr_lock_thread_list;                     /* List of threads in use */
99
 
ulong max_write_lock_count= ~(ulong) 0L;
 
96
static list<THR_LOCK *> thr_lock_thread_list;          /* List of threads in use */
 
97
 
 
98
uint64_t max_write_lock_count= ~(uint64_t) 0L;
100
99
 
101
100
static inline pthread_cond_t *get_cond(void)
102
101
{
109
108
 
110
109
bool init_thr_lock()
111
110
{
112
 
  thr_lock_inited=1;
113
 
  return 0;
 
111
  thr_lock_inited= true;
 
112
 
 
113
  return false;
114
114
}
115
115
 
116
116
static inline bool
119
119
  return rhs == lhs;
120
120
}
121
121
 
122
 
#ifdef EXTRA_DEBUG
123
 
 
124
 
#define MAX_THREADS 100
125
 
#define MAX_LOCKS   100
126
 
#define MAX_FOUND_ERRORS        10              /* Report 10 first errors */
127
 
static uint32_t found_errors=0;
128
 
 
129
 
static int check_lock(struct st_lock_list *list, const char* lock_type,
130
 
                      const char *where, bool same_owner, bool no_cond)
131
 
{
132
 
  THR_LOCK_DATA *data,**prev;
133
 
  uint32_t count=0;
134
 
  THR_LOCK_OWNER *first_owner;
135
 
 
136
 
  prev= &list->data;
137
 
  if (list->data)
138
 
  {
139
 
    enum thr_lock_type last_lock_type=list->data->type;
140
 
 
141
 
    if (same_owner && list->data)
142
 
      first_owner= list->data->owner;
143
 
    for (data=list->data; data && count++ < MAX_LOCKS ; data=data->next)
144
 
    {
145
 
      if (data->type != last_lock_type)
146
 
        last_lock_type=TL_IGNORE;
147
 
      if (data->prev != prev)
148
 
      {
149
 
        fprintf(stderr,
150
 
                "Warning: prev link %d didn't point at previous lock at %s: %s\n",
151
 
                count, lock_type, where);
152
 
        return 1;
153
 
      }
154
 
      if (same_owner &&
155
 
          !thr_lock_owner_equal(data->owner, first_owner) &&
156
 
          last_lock_type != TL_WRITE_ALLOW_WRITE)
157
 
      {
158
 
        fprintf(stderr,
159
 
                "Warning: Found locks from different threads in %s: %s\n",
160
 
                lock_type,where);
161
 
        return 1;
162
 
      }
163
 
      if (no_cond && data->cond)
164
 
      {
165
 
        fprintf(stderr,
166
 
                "Warning: Found active lock with not reset cond %s: %s\n",
167
 
                lock_type,where);
168
 
        return 1;
169
 
      }
170
 
      prev= &data->next;
171
 
    }
172
 
    if (data)
173
 
    {
174
 
      fprintf(stderr,"Warning: found too many locks at %s: %s\n",
175
 
              lock_type,where);
176
 
      return 1;
177
 
    }
178
 
  }
179
 
  if (prev != list->last)
180
 
  {
181
 
    fprintf(stderr,"Warning: last didn't point at last lock at %s: %s\n",
182
 
            lock_type, where);
183
 
    return 1;
184
 
  }
185
 
  return 0;
186
 
}
187
 
 
188
 
 
189
 
static void check_locks(THR_LOCK *lock, const char *where,
190
 
                        bool allow_no_locks)
191
 
{
192
 
  uint32_t old_found_errors=found_errors;
193
 
 
194
 
  if (found_errors < MAX_FOUND_ERRORS)
195
 
  {
196
 
    if (check_lock(&lock->write,"write",where,1,1) |
197
 
        check_lock(&lock->write_wait,"write_wait",where,0,0) |
198
 
        check_lock(&lock->read,"read",where,0,1) |
199
 
        check_lock(&lock->read_wait,"read_wait",where,0,0))
200
 
      found_errors++;
201
 
 
202
 
    if (found_errors < MAX_FOUND_ERRORS)
203
 
    {
204
 
      uint32_t count=0;
205
 
      THR_LOCK_DATA *data;
206
 
      for (data=lock->read.data ; data ; data=data->next)
207
 
      {
208
 
        if ((int) data->type == (int) TL_READ_NO_INSERT)
209
 
          count++;
210
 
        /* Protect against infinite loop. */
211
 
        assert(count <= lock->read_no_write_count);
212
 
      }
213
 
      if (count != lock->read_no_write_count)
214
 
      {
215
 
        found_errors++;
216
 
        fprintf(stderr,
217
 
                "Warning at '%s': Locks read_no_write_count was %u when it should have been %u\n", where, lock->read_no_write_count,count);
218
 
      }      
219
 
 
220
 
      if (!lock->write.data)
221
 
      {
222
 
        if (!allow_no_locks && !lock->read.data &&
223
 
            (lock->write_wait.data || lock->read_wait.data))
224
 
        {
225
 
          found_errors++;
226
 
          fprintf(stderr,
227
 
                  "Warning at '%s': No locks in use but locks are in wait queue\n",
228
 
                  where);
229
 
        }
230
 
        if (!lock->write_wait.data)
231
 
        {
232
 
          if (!allow_no_locks && lock->read_wait.data)
233
 
          {
234
 
            found_errors++;
235
 
            fprintf(stderr,
236
 
                    "Warning at '%s': No write locks and waiting read locks\n",
237
 
                    where);
238
 
          }
239
 
        }
240
 
        else
241
 
        {
242
 
          if (!allow_no_locks &&
243
 
              (((lock->write_wait.data->type == TL_WRITE_CONCURRENT_INSERT ||
244
 
                 lock->write_wait.data->type == TL_WRITE_ALLOW_WRITE) &&
245
 
                !lock->read_no_write_count) ||
246
 
               lock->write_wait.data->type == TL_WRITE_ALLOW_READ ||
247
 
               (lock->write_wait.data->type == TL_WRITE_DELAYED &&
248
 
                !lock->read.data)))
249
 
          {
250
 
            found_errors++;
251
 
            fprintf(stderr,
252
 
                    "Warning at '%s': Write lock %d waiting while no exclusive read locks\n",where,(int) lock->write_wait.data->type);
253
 
          }
254
 
        }             
255
 
      }
256
 
      else
257
 
      {                                         /* Have write lock */
258
 
        if (lock->write_wait.data)
259
 
        {
260
 
          if (!allow_no_locks && 
261
 
              lock->write.data->type == TL_WRITE_ALLOW_WRITE &&
262
 
              lock->write_wait.data->type == TL_WRITE_ALLOW_WRITE)
263
 
          {
264
 
            found_errors++;
265
 
            fprintf(stderr,
266
 
                    "Warning at '%s': Found WRITE_ALLOW_WRITE lock waiting for WRITE_ALLOW_WRITE lock\n",
267
 
                    where);
268
 
          }
269
 
        }
270
 
        if (lock->read.data)
271
 
        {
272
 
          if (!thr_lock_owner_equal(lock->write.data->owner,
273
 
                                    lock->read.data->owner) &&
274
 
              ((lock->write.data->type > TL_WRITE_DELAYED &&
275
 
                lock->write.data->type != TL_WRITE_ONLY) ||
276
 
               ((lock->write.data->type == TL_WRITE_CONCURRENT_INSERT ||
277
 
                 lock->write.data->type == TL_WRITE_ALLOW_WRITE) &&
278
 
                lock->read_no_write_count)))
279
 
          {
280
 
            found_errors++;
281
 
            fprintf(stderr,
282
 
                    "Warning at '%s': Found lock of type %d that is write and read locked\n",
283
 
                    where, lock->write.data->type);
284
 
          }
285
 
        }
286
 
        if (lock->read_wait.data)
287
 
        {
288
 
          if (!allow_no_locks && lock->write.data->type <= TL_WRITE_DELAYED &&
289
 
              lock->read_wait.data->type <= TL_READ_HIGH_PRIORITY)
290
 
          {
291
 
            found_errors++;
292
 
            fprintf(stderr,
293
 
                    "Warning at '%s': Found read lock of type %d waiting for write lock of type %d\n",
294
 
                    where,
295
 
                    (int) lock->read_wait.data->type,
296
 
                    (int) lock->write.data->type);
297
 
          }
298
 
        }
299
 
      }
300
 
    }
301
 
  }
302
 
  return;
303
 
}
304
 
 
305
 
#else /* EXTRA_DEBUG */
306
 
#define check_locks(A,B,C)
307
 
#endif
308
 
 
309
122
 
310
123
        /* Initialize a lock */
311
124
 
319
132
  lock->write.last= &lock->write.data;
320
133
 
321
134
  pthread_mutex_lock(&THR_LOCK_lock);           /* Add to locks in use */
322
 
  lock->list.data=(void*) lock;
323
 
  thr_lock_thread_list=list_add(thr_lock_thread_list,&lock->list);
 
135
  thr_lock_thread_list.push_front(lock);
324
136
  pthread_mutex_unlock(&THR_LOCK_lock);
325
 
  return;
326
137
}
327
138
 
328
139
 
330
141
{
331
142
  pthread_mutex_destroy(&lock->mutex);
332
143
  pthread_mutex_lock(&THR_LOCK_lock);
333
 
  thr_lock_thread_list=list_delete(thr_lock_thread_list,&lock->list);
 
144
  thr_lock_thread_list.remove(lock);
334
145
  pthread_mutex_unlock(&THR_LOCK_lock);
335
 
  return;
336
146
}
337
147
 
338
148
 
339
149
void thr_lock_info_init(THR_LOCK_INFO *info)
340
150
{
341
151
  struct st_my_thread_var *tmp= my_thread_var;
342
 
  info->thread=    tmp->pthread_self;
 
152
  info->thread= tmp->pthread_self;
343
153
  info->thread_id= tmp->id;
344
154
  info->n_cursors= 0;
345
155
}
348
158
 
349
159
void thr_lock_data_init(THR_LOCK *lock,THR_LOCK_DATA *data, void *param)
350
160
{
351
 
  data->lock=lock;
352
 
  data->type=TL_UNLOCK;
353
 
  data->owner= 0;                               /* no owner yet */
354
 
  data->status_param=param;
355
 
  data->cond=0;
 
161
  data->lock= lock;
 
162
  data->type= TL_UNLOCK;
 
163
  data->owner= NULL;                               /* no owner yet */
 
164
  data->status_param= param;
 
165
  data->cond= NULL;
356
166
}
357
167
 
358
168
 
362
172
  for ( ; data ; data=data->next)
363
173
  {
364
174
    if (thr_lock_owner_equal(data->owner, owner))
365
 
      return 1;                                 /* Already locked by thread */
366
 
  }
367
 
  return 0;
368
 
}
369
 
 
370
 
static inline bool have_specific_lock(THR_LOCK_DATA *data,
371
 
                                         enum thr_lock_type type)
372
 
{
373
 
  for ( ; data ; data=data->next)
374
 
  {
375
 
    if (data->type == type)
376
 
      return 1;
377
 
  }
378
 
  return 0;
379
 
}
380
 
 
 
175
      return true;                                      /* Already locked by thread */
 
176
  }
 
177
  return false;
 
178
}
381
179
 
382
180
static void wake_up_waiters(THR_LOCK *lock);
383
181
 
433
231
    }
434
232
    if (rc == ETIMEDOUT || rc == ETIME)
435
233
    {
436
 
      /* purecov: begin inspected */
437
234
      result= THR_LOCK_WAIT_TIMEOUT;
438
235
      break;
439
 
      /* purecov: end */
440
236
    }
441
237
  }
442
238
  if (data->cond || data->type == TL_UNLOCK)
448
244
      else
449
245
        wait->last=data->prev;
450
246
      data->type= TL_UNLOCK;                    /* No lock */
451
 
      check_locks(data->lock, "killed or timed out wait_for_lock", 1);
452
247
      wake_up_waiters(data->lock);
453
248
    }
454
 
    else
455
 
    {
456
 
      check_locks(data->lock, "aborted wait_for_lock", 0);
457
 
    }
458
249
  }
459
250
  else
460
251
  {
461
252
    result= THR_LOCK_SUCCESS;
462
253
    if (data->lock->get_status)
463
254
      (*data->lock->get_status)(data->status_param, 0);
464
 
    check_locks(data->lock,"got wait_for_lock",0);
465
255
  }
466
256
  pthread_mutex_unlock(&data->lock->mutex);
467
257
 
468
258
  /* The following must be done after unlock of lock->mutex */
469
259
  pthread_mutex_lock(&thread_var->mutex);
470
 
  thread_var->current_mutex= 0;
471
 
  thread_var->current_cond=  0;
 
260
  thread_var->current_mutex= NULL;
 
261
  thread_var->current_cond= NULL;
472
262
  pthread_mutex_unlock(&thread_var->mutex);
473
263
  return(result);
474
264
}
488
278
  data->type=lock_type;
489
279
  data->owner= owner;                           /* Must be reset ! */
490
280
  pthread_mutex_lock(&lock->mutex);
491
 
  check_locks(lock,(uint) lock_type <= (uint) TL_READ_NO_INSERT ?
492
 
              "enter read_lock" : "enter write_lock",0);
493
281
  if ((int) lock_type <= (int) TL_READ_NO_INSERT)
494
282
  {
495
283
    /* Request for READ lock */
505
293
      */
506
294
 
507
295
      if (thr_lock_owner_equal(data->owner, lock->write.data->owner) ||
508
 
          (lock->write.data->type <= TL_WRITE_DELAYED &&
509
 
           (((int) lock_type <= (int) TL_READ_HIGH_PRIORITY) ||
 
296
          (lock->write.data->type <= TL_WRITE_CONCURRENT_INSERT &&
 
297
           (((int) lock_type <= (int) TL_READ_WITH_SHARED_LOCKS) ||
510
298
            (lock->write.data->type != TL_WRITE_CONCURRENT_INSERT &&
511
299
             lock->write.data->type != TL_WRITE_ALLOW_READ))))
512
300
      {                                         /* Already got a write lock */
515
303
        lock->read.last= &data->next;
516
304
        if (lock_type == TL_READ_NO_INSERT)
517
305
          lock->read_no_write_count++;
518
 
        check_locks(lock,"read lock with old write lock",0);
519
306
        if (lock->get_status)
520
307
          (*lock->get_status)(data->status_param, 0);
521
308
        statistic_increment(locks_immediate,&THR_LOCK_lock);
530
317
      }
531
318
    }
532
319
    else if (!lock->write_wait.data ||
533
 
             lock->write_wait.data->type <= TL_WRITE_LOW_PRIORITY ||
534
 
             lock_type == TL_READ_HIGH_PRIORITY ||
 
320
             lock->write_wait.data->type <= TL_WRITE_DEFAULT ||
535
321
             have_old_read_lock(lock->read.data, data->owner))
536
322
    {                                           /* No important write-locks */
537
323
      (*lock->read.last)=data;                  /* Add to running FIFO */
541
327
        (*lock->get_status)(data->status_param, 0);
542
328
      if (lock_type == TL_READ_NO_INSERT)
543
329
        lock->read_no_write_count++;
544
 
      check_locks(lock,"read lock with no write locks",0);
545
330
      statistic_increment(locks_immediate,&THR_LOCK_lock);
546
331
      goto end;
547
332
    }
554
339
  }
555
340
  else                                          /* Request for WRITE lock */
556
341
  {
557
 
    if (lock_type == TL_WRITE_DELAYED)
558
 
    {
559
 
      if (lock->write.data && lock->write.data->type == TL_WRITE_ONLY)
560
 
      {
561
 
        data->type=TL_UNLOCK;
562
 
        result= THR_LOCK_ABORTED;               /* Can't wait for this one */
563
 
        goto end;
564
 
      }
565
 
      /*
566
 
        if there is a TL_WRITE_ALLOW_READ lock, we have to wait for a lock
567
 
        (TL_WRITE_ALLOW_READ is used for ALTER TABLE in MySQL)
568
 
      */
569
 
      if ((!lock->write.data ||
570
 
           lock->write.data->type != TL_WRITE_ALLOW_READ) &&
571
 
          !have_specific_lock(lock->write_wait.data,TL_WRITE_ALLOW_READ) &&
572
 
          (lock->write.data || lock->read.data))
573
 
      {
574
 
        /* Add delayed write lock to write_wait queue, and return at once */
575
 
        (*lock->write_wait.last)=data;
576
 
        data->prev=lock->write_wait.last;
577
 
        lock->write_wait.last= &data->next;
578
 
        data->cond=get_cond();
579
 
        /*
580
 
          We don't have to do get_status here as we will do it when we change
581
 
          the delayed lock to a real write lock
582
 
        */
583
 
        statistic_increment(locks_immediate,&THR_LOCK_lock);
584
 
        goto end;
585
 
      }
586
 
    }
587
 
    else if (lock_type == TL_WRITE_CONCURRENT_INSERT && ! lock->check_status)
 
342
    if (lock_type == TL_WRITE_CONCURRENT_INSERT && ! lock->check_status)
588
343
      data->type=lock_type= thr_upgraded_concurrent_insert_lock;
589
344
 
590
345
    if (lock->write.data)                       /* If there is a write lock */
619
374
        (*lock->write.last)=data;       /* Add to running fifo */
620
375
        data->prev=lock->write.last;
621
376
        lock->write.last= &data->next;
622
 
        check_locks(lock,"second write lock",0);
623
377
        if (data->lock->get_status)
624
378
          (*data->lock->get_status)(data->status_param, 0);
625
379
        statistic_increment(locks_immediate,&THR_LOCK_lock);
642
396
        }
643
397
 
644
398
        if (!lock->read.data ||
645
 
            (lock_type <= TL_WRITE_DELAYED &&
 
399
            (lock_type <= TL_WRITE_CONCURRENT_INSERT &&
646
400
             ((lock_type != TL_WRITE_CONCURRENT_INSERT &&
647
401
               lock_type != TL_WRITE_ALLOW_WRITE) ||
648
402
              !lock->read_no_write_count)))
652
406
          lock->write.last= &data->next;
653
407
          if (data->lock->get_status)
654
408
            (*data->lock->get_status)(data->status_param, concurrent_insert);
655
 
          check_locks(lock,"only write lock",0);
656
409
          statistic_increment(locks_immediate,&THR_LOCK_lock);
657
410
          goto end;
658
411
        }
680
433
}
681
434
 
682
435
 
683
 
static inline void free_all_read_locks(THR_LOCK *lock,
684
 
                                       bool using_concurrent_insert)
 
436
static void free_all_read_locks(THR_LOCK *lock, bool using_concurrent_insert)
685
437
{
686
438
  THR_LOCK_DATA *data=lock->read_wait.data;
687
439
 
688
 
  check_locks(lock,"before freeing read locks",1);
689
 
 
690
440
  /* move all locks from read_wait list to read list */
691
441
  (*lock->read.last)=data;
692
442
  data->prev=lock->read.last;
703
453
      if (using_concurrent_insert)
704
454
      {
705
455
        /*
706
 
          We can't free this lock; 
 
456
          We can't free this lock;
707
457
          Link lock away from read chain back into read_wait chain
708
458
        */
709
459
        if (((*data->prev)=data->next))
716
466
        continue;
717
467
      }
718
468
      lock->read_no_write_count++;
719
 
    }      
 
469
    }
720
470
    data->cond=0;                               /* Mark thread free */
721
471
    pthread_cond_signal(cond);
722
472
  } while ((data=data->next));
723
473
  *lock->read_wait.last=0;
724
474
  if (!lock->read_wait.data)
725
475
    lock->write_lock_count=0;
726
 
  check_locks(lock,"after giving read locks",0);
727
476
}
728
477
 
729
478
        /* Unlock lock and free next thread on same lock */
733
482
  THR_LOCK *lock=data->lock;
734
483
  enum thr_lock_type lock_type=data->type;
735
484
  pthread_mutex_lock(&lock->mutex);
736
 
  check_locks(lock,"start of release lock",0);
737
485
 
738
486
  if (((*data->prev)=data->next))               /* remove from lock-list */
739
487
    data->next->prev= data->prev;
740
488
  else if (lock_type <= TL_READ_NO_INSERT)
741
489
    lock->read.last=data->prev;
742
 
  else if (lock_type == TL_WRITE_DELAYED && data->cond)
743
 
  {
744
 
    /*
745
 
      This only happens in extreme circumstances when a 
746
 
      write delayed lock that is waiting for a lock
747
 
    */
748
 
    lock->write_wait.last=data->prev;           /* Put it on wait queue */
749
 
  }
750
490
  else
751
491
    lock->write.last=data->prev;
752
492
  if (lock_type >= TL_WRITE_CONCURRENT_INSERT)
762
502
  if (lock_type == TL_READ_NO_INSERT)
763
503
    lock->read_no_write_count--;
764
504
  data->type=TL_UNLOCK;                         /* Mark unlocked */
765
 
  check_locks(lock,"after releasing lock",1);
766
505
  wake_up_waiters(lock);
767
506
  pthread_mutex_unlock(&lock->mutex);
768
507
  return;
789
528
    {
790
529
      /* Release write-locks with TL_WRITE or TL_WRITE_ONLY priority first */
791
530
      if (data &&
792
 
          (data->type != TL_WRITE_LOW_PRIORITY || !lock->read_wait.data ||
793
 
           lock->read_wait.data->type < TL_READ_HIGH_PRIORITY))
 
531
          (!lock->read_wait.data || lock->read_wait.data->type <= TL_READ_WITH_SHARED_LOCKS))
794
532
      {
795
533
        if (lock->write_lock_count++ > max_write_lock_count)
796
534
        {
826
564
            break;
827
565
          data=lock->write_wait.data;           /* Free this too */
828
566
        }
829
 
        if (data->type >= TL_WRITE_LOW_PRIORITY)
 
567
        if (data->type >= TL_WRITE)
830
568
          goto end;
831
569
        /* Release possible read locks together with the write lock */
832
570
      }
837
575
                             data->type == TL_WRITE_ALLOW_WRITE));
838
576
    }
839
577
    else if (data &&
840
 
             (lock_type=data->type) <= TL_WRITE_DELAYED &&
 
578
             (lock_type=data->type) <= TL_WRITE_CONCURRENT_INSERT &&
841
579
             ((lock_type != TL_WRITE_CONCURRENT_INSERT &&
842
580
               lock_type != TL_WRITE_ALLOW_WRITE) ||
843
581
              !lock->read_no_write_count))
878
616
      free_all_read_locks(lock,0);
879
617
  }
880
618
end:
881
 
  check_locks(lock, "after waking up waiters", 0);
882
619
  return;
883
620
}
884
621
 
890
627
*/
891
628
 
892
629
 
893
 
#define LOCK_CMP(A,B) ((unsigned char*) (A->lock) - (uint) ((A)->type) < (unsigned char*) (B->lock)- (uint) ((B)->type))
 
630
#define LOCK_CMP(A,B) ((unsigned char*) (A->lock) - (uint32_t) ((A)->type) < (unsigned char*) (B->lock)- (uint32_t) ((B)->type))
894
631
 
895
632
static void sort_locks(THR_LOCK_DATA **data,uint32_t count)
896
633
{
925
662
    enum enum_thr_lock_result result= thr_lock(*pos, owner, (*pos)->type);
926
663
    if (result != THR_LOCK_SUCCESS)
927
664
    {                                           /* Aborted */
928
 
      thr_multi_unlock(data,(uint) (pos-data));
 
665
      thr_multi_unlock(data,(uint32_t) (pos-data));
929
666
      return(result);
930
667
    }
931
 
#ifdef MAIN
932
 
    printf("Thread: %s  Got lock: 0x%lx  type: %d\n",my_thread_name(),
933
 
           (long) pos[0]->lock, pos[0]->type); fflush(stdout);
934
 
#endif
935
668
  }
936
669
  /*
937
670
    Ensure that all get_locks() have the same status
991
724
 
992
725
  for (pos=data,end=data+count; pos < end ; pos++)
993
726
  {
994
 
#ifdef MAIN
995
 
    printf("Thread: %s  Rel lock: 0x%lx  type: %d\n",
996
 
           my_thread_name(), (long) pos[0]->lock, pos[0]->type);
997
 
    fflush(stdout);
998
 
#endif
999
727
    if ((*pos)->type != TL_UNLOCK)
1000
728
      thr_unlock(*pos);
1001
729
  }
1007
735
  TL_WRITE_ONLY to abort any new accesses to the lock
1008
736
*/
1009
737
 
1010
 
void thr_abort_locks(THR_LOCK *lock, bool upgrade_lock)
 
738
void thr_abort_locks(THR_LOCK *lock)
1011
739
{
1012
740
  THR_LOCK_DATA *data;
1013
741
  pthread_mutex_lock(&lock->mutex);
1014
742
 
1015
743
  for (data=lock->read_wait.data; data ; data=data->next)
1016
744
  {
1017
 
    data->type=TL_UNLOCK;                       /* Mark killed */
 
745
    data->type= TL_UNLOCK;                      /* Mark killed */
1018
746
    /* It's safe to signal the cond first: we're still holding the mutex. */
1019
747
    pthread_cond_signal(data->cond);
1020
 
    data->cond=0;                               /* Removed from list */
 
748
    data->cond= NULL;                           /* Removed from list */
1021
749
  }
1022
750
  for (data=lock->write_wait.data; data ; data=data->next)
1023
751
  {
1024
752
    data->type=TL_UNLOCK;
1025
753
    pthread_cond_signal(data->cond);
1026
 
    data->cond=0;
 
754
    data->cond= NULL;
1027
755
  }
1028
756
  lock->read_wait.last= &lock->read_wait.data;
1029
757
  lock->write_wait.last= &lock->write_wait.data;
1030
758
  lock->read_wait.data=lock->write_wait.data=0;
1031
 
  if (upgrade_lock && lock->write.data)
 
759
  if (lock->write.data)
1032
760
    lock->write.data->type=TL_WRITE_ONLY;
1033
761
  pthread_mutex_unlock(&lock->mutex);
1034
762
  return;
1049
777
  pthread_mutex_lock(&lock->mutex);
1050
778
  for (data= lock->read_wait.data; data ; data= data->next)
1051
779
  {
1052
 
    if (data->owner->info->thread_id == thread_id)    /* purecov: tested */
 
780
    if (data->owner->info->thread_id == thread_id)
1053
781
    {
1054
782
      data->type= TL_UNLOCK;                    /* Mark killed */
1055
783
      /* It's safe to signal the cond first: we're still holding the mutex. */
1065
793
  }
1066
794
  for (data= lock->write_wait.data; data ; data= data->next)
1067
795
  {
1068
 
    if (data->owner->info->thread_id == thread_id) /* purecov: tested */
 
796
    if (data->owner->info->thread_id == thread_id)
1069
797
    {
1070
798
      data->type= TL_UNLOCK;
1071
799
      found= true;
1072
800
      pthread_cond_signal(data->cond);
1073
 
      data->cond= 0;
 
801
      data->cond= NULL;
1074
802
 
1075
803
      if (((*data->prev)= data->next))
1076
804
        data->next->prev= data->prev;
1082
810
  pthread_mutex_unlock(&lock->mutex);
1083
811
  return(found);
1084
812
}
1085
 
 
1086
 
 
1087
 
/*
1088
 
  Downgrade a WRITE_* to a lower WRITE level
1089
 
  SYNOPSIS
1090
 
    thr_downgrade_write_lock()
1091
 
    in_data                   Lock data of thread downgrading its lock
1092
 
    new_lock_type             New write lock type
1093
 
  RETURN VALUE
1094
 
    NONE
1095
 
  DESCRIPTION
1096
 
    This can be used to downgrade a lock already owned. When the downgrade
1097
 
    occurs also other waiters, both readers and writers can be allowed to
1098
 
    start.
1099
 
    The previous lock is often TL_WRITE_ONLY but can also be
1100
 
    TL_WRITE and TL_WRITE_ALLOW_READ. The normal downgrade variants are
1101
 
    TL_WRITE_ONLY => TL_WRITE_ALLOW_READ After a short exclusive lock
1102
 
    TL_WRITE_ALLOW_READ => TL_WRITE_ALLOW_WRITE After discovering that the
1103
 
    operation didn't need such a high lock.
1104
 
    TL_WRITE_ONLY => TL_WRITE after a short exclusive lock while holding a
1105
 
    write table lock
1106
 
    TL_WRITE_ONLY => TL_WRITE_ALLOW_WRITE After a short exclusive lock after
1107
 
    already earlier having dongraded lock to TL_WRITE_ALLOW_WRITE
1108
 
    The implementation is conservative and rather don't start rather than
1109
 
    go on unknown paths to start, the common cases are handled.
1110
 
 
1111
 
    NOTE:
1112
 
    In its current implementation it is only allowed to downgrade from
1113
 
    TL_WRITE_ONLY. In this case there are no waiters. Thus no wake up
1114
 
    logic is required.
1115
 
*/
1116
 
 
1117
 
void thr_downgrade_write_lock(THR_LOCK_DATA *in_data,
1118
 
                              enum thr_lock_type new_lock_type)
1119
 
{
1120
 
  THR_LOCK *lock=in_data->lock;
1121
 
 
1122
 
  pthread_mutex_lock(&lock->mutex);
1123
 
  in_data->type= new_lock_type;
1124
 
  check_locks(lock,"after downgrading lock",0);
1125
 
 
1126
 
  pthread_mutex_unlock(&lock->mutex);
1127
 
  return;
1128
 
}
1129
 
 
1130
 
/* Upgrade a WRITE_DELAY lock to a WRITE_LOCK */
1131
 
 
1132
 
bool thr_upgrade_write_delay_lock(THR_LOCK_DATA *data)
1133
 
{
1134
 
  THR_LOCK *lock=data->lock;
1135
 
 
1136
 
  pthread_mutex_lock(&lock->mutex);
1137
 
  if (data->type == TL_UNLOCK || data->type >= TL_WRITE_LOW_PRIORITY)
1138
 
  {
1139
 
    pthread_mutex_unlock(&lock->mutex);
1140
 
    return(data->type == TL_UNLOCK);    /* Test if Aborted */
1141
 
  }
1142
 
  check_locks(lock,"before upgrading lock",0);
1143
 
  /* TODO:  Upgrade to TL_WRITE_CONCURRENT_INSERT in some cases */
1144
 
  data->type=TL_WRITE;                          /* Upgrade lock */
1145
 
 
1146
 
  /* Check if someone has given us the lock */
1147
 
  if (!data->cond)
1148
 
  {
1149
 
    if (!lock->read.data)                       /* No read locks */
1150
 
    {                                           /* We have the lock */
1151
 
      if (data->lock->get_status)
1152
 
        (*data->lock->get_status)(data->status_param, 0);
1153
 
      pthread_mutex_unlock(&lock->mutex);
1154
 
      return(0);
1155
 
    }
1156
 
 
1157
 
    if (((*data->prev)=data->next))             /* remove from lock-list */
1158
 
      data->next->prev= data->prev;
1159
 
    else
1160
 
      lock->write.last=data->prev;
1161
 
 
1162
 
    if ((data->next=lock->write_wait.data))     /* Put first in lock_list */
1163
 
      data->next->prev= &data->next;
1164
 
    else
1165
 
      lock->write_wait.last= &data->next;
1166
 
    data->prev= &lock->write_wait.data;
1167
 
    lock->write_wait.data=data;
1168
 
    check_locks(lock,"upgrading lock",0);
1169
 
  }
1170
 
  else
1171
 
  {
1172
 
    check_locks(lock,"waiting for lock",0);
1173
 
  }
1174
 
  return(wait_for_lock(&lock->write_wait,data,1));
1175
 
}
1176
 
 
1177
 
 
1178
 
/* downgrade a WRITE lock to a WRITE_DELAY lock if there is pending locks */
1179
 
 
1180
 
bool thr_reschedule_write_lock(THR_LOCK_DATA *data)
1181
 
{
1182
 
  THR_LOCK *lock=data->lock;
1183
 
 
1184
 
  pthread_mutex_lock(&lock->mutex);
1185
 
  if (!lock->read_wait.data)                    /* No waiting read locks */
1186
 
  {
1187
 
    pthread_mutex_unlock(&lock->mutex);
1188
 
    return(0);
1189
 
  }
1190
 
 
1191
 
  data->type=TL_WRITE_DELAYED;
1192
 
  if (lock->update_status)
1193
 
    (*lock->update_status)(data->status_param);
1194
 
  if (((*data->prev)=data->next))               /* remove from lock-list */
1195
 
    data->next->prev= data->prev;
1196
 
  else
1197
 
    lock->write.last=data->prev;
1198
 
 
1199
 
  if ((data->next=lock->write_wait.data))       /* Put first in lock_list */
1200
 
    data->next->prev= &data->next;
1201
 
  else
1202
 
    lock->write_wait.last= &data->next;
1203
 
  data->prev= &lock->write_wait.data;
1204
 
  data->cond=get_cond();                        /* This was zero */
1205
 
  lock->write_wait.data=data;
1206
 
  free_all_read_locks(lock,0);
1207
 
 
1208
 
  pthread_mutex_unlock(&lock->mutex);
1209
 
  return(thr_upgrade_write_delay_lock(data));
1210
 
}
1211
 
 
1212
 
 
1213
 
#include <my_sys.h>
1214
 
 
1215
 
/*****************************************************************************
1216
 
** Test of thread locks
1217
 
****************************************************************************/
1218
 
 
1219
 
#ifdef MAIN
1220
 
 
1221
 
struct st_test {
1222
 
  uint32_t lock_nr;
1223
 
  enum thr_lock_type lock_type;
1224
 
};
1225
 
 
1226
 
THR_LOCK locks[5];                      /* 4 locks */
1227
 
 
1228
 
struct st_test test_0[] = {{0,TL_READ}};        /* One lock */
1229
 
struct st_test test_1[] = {{0,TL_READ},{0,TL_WRITE}}; /* Read and write lock of lock 0 */
1230
 
struct st_test test_2[] = {{1,TL_WRITE},{0,TL_READ},{2,TL_READ}};
1231
 
struct st_test test_3[] = {{2,TL_WRITE},{1,TL_READ},{0,TL_READ}}; /* Deadlock with test_2 ? */
1232
 
struct st_test test_4[] = {{0,TL_WRITE},{0,TL_READ},{0,TL_WRITE},{0,TL_READ}};
1233
 
struct st_test test_5[] = {{0,TL_READ},{1,TL_READ},{2,TL_READ},{3,TL_READ}}; /* Many reads */
1234
 
struct st_test test_6[] = {{0,TL_WRITE},{1,TL_WRITE},{2,TL_WRITE},{3,TL_WRITE}}; /* Many writes */
1235
 
struct st_test test_7[] = {{3,TL_READ}};
1236
 
struct st_test test_8[] = {{1,TL_READ_NO_INSERT},{2,TL_READ_NO_INSERT},{3,TL_READ_NO_INSERT}};  /* Should be quick */
1237
 
struct st_test test_9[] = {{4,TL_READ_HIGH_PRIORITY}};
1238
 
struct st_test test_10[] ={{4,TL_WRITE}};
1239
 
struct st_test test_11[] = {{0,TL_WRITE_LOW_PRIORITY},{1,TL_WRITE_LOW_PRIORITY},{2,TL_WRITE_LOW_PRIORITY},{3,TL_WRITE_LOW_PRIORITY}}; /* Many writes */
1240
 
struct st_test test_12[] = {{0,TL_WRITE_ALLOW_READ},{1,TL_WRITE_ALLOW_READ},{2,TL_WRITE_ALLOW_READ},{3,TL_WRITE_ALLOW_READ}}; /* Many writes */
1241
 
struct st_test test_13[] = {{0,TL_WRITE_CONCURRENT_INSERT},{1,TL_WRITE_CONCURRENT_INSERT},{2,TL_WRITE_CONCURRENT_INSERT},{3,TL_WRITE_CONCURRENT_INSERT}};
1242
 
struct st_test test_14[] = {{0,TL_WRITE_CONCURRENT_INSERT},{1,TL_READ}};
1243
 
struct st_test test_15[] = {{0,TL_WRITE_ALLOW_WRITE},{1,TL_READ}};
1244
 
struct st_test test_16[] = {{0,TL_WRITE_ALLOW_WRITE},{1,TL_WRITE_ALLOW_WRITE}};
1245
 
 
1246
 
struct st_test *tests[] = {test_0,test_1,test_2,test_3,test_4,test_5,test_6,
1247
 
                           test_7,test_8,test_9,test_10,test_11,test_12,
1248
 
                           test_13,test_14,test_15,test_16};
1249
 
int lock_counts[]= {sizeof(test_0)/sizeof(struct st_test),
1250
 
                    sizeof(test_1)/sizeof(struct st_test),
1251
 
                    sizeof(test_2)/sizeof(struct st_test),
1252
 
                    sizeof(test_3)/sizeof(struct st_test),
1253
 
                    sizeof(test_4)/sizeof(struct st_test),
1254
 
                    sizeof(test_5)/sizeof(struct st_test),
1255
 
                    sizeof(test_6)/sizeof(struct st_test),
1256
 
                    sizeof(test_7)/sizeof(struct st_test),
1257
 
                    sizeof(test_8)/sizeof(struct st_test),
1258
 
                    sizeof(test_9)/sizeof(struct st_test),
1259
 
                    sizeof(test_10)/sizeof(struct st_test),
1260
 
                    sizeof(test_11)/sizeof(struct st_test),
1261
 
                    sizeof(test_12)/sizeof(struct st_test),
1262
 
                    sizeof(test_13)/sizeof(struct st_test),
1263
 
                    sizeof(test_14)/sizeof(struct st_test),
1264
 
                    sizeof(test_15)/sizeof(struct st_test),
1265
 
                    sizeof(test_16)/sizeof(struct st_test)
1266
 
};
1267
 
 
1268
 
 
1269
 
static pthread_cond_t COND_thread_count;
1270
 
static pthread_mutex_t LOCK_thread_count;
1271
 
static uint32_t thread_count;
1272
 
static uint32_t sum=0;
1273
 
 
1274
 
#define MAX_LOCK_COUNT 8
1275
 
 
1276
 
/* The following functions is for WRITE_CONCURRENT_INSERT */
1277
 
 
1278
 
static void test_get_status(void* param __attribute__((unused)),
1279
 
                            int concurrent_insert __attribute__((unused)))
1280
 
{
1281
 
}
1282
 
 
1283
 
static void test_update_status(void* param __attribute__((unused)))
1284
 
{
1285
 
}
1286
 
 
1287
 
static void test_copy_status(void* to __attribute__((unused)) ,
1288
 
                             void *from __attribute__((unused)))
1289
 
{
1290
 
}
1291
 
 
1292
 
static bool test_check_status(void* param __attribute__((unused)))
1293
 
{
1294
 
  return 0;
1295
 
}
1296
 
 
1297
 
 
1298
 
static void *test_thread(void *arg)
1299
 
{
1300
 
  int i,j,param=*((int*) arg);
1301
 
  THR_LOCK_DATA data[MAX_LOCK_COUNT];
1302
 
  THR_LOCK_OWNER owner;
1303
 
  THR_LOCK_INFO lock_info;
1304
 
  THR_LOCK_DATA *multi_locks[MAX_LOCK_COUNT];
1305
 
  my_thread_init();
1306
 
 
1307
 
  printf("Thread %s (%d) started\n",my_thread_name(),param); fflush(stdout);
1308
 
 
1309
 
 
1310
 
  thr_lock_info_init(&lock_info);
1311
 
  thr_lock_owner_init(&owner, &lock_info);
1312
 
  for (i=0; i < lock_counts[param] ; i++)
1313
 
    thr_lock_data_init(locks+tests[param][i].lock_nr,data+i,NULL);
1314
 
  for (j=1 ; j < 10 ; j++)              /* try locking 10 times */
1315
 
  {
1316
 
    for (i=0; i < lock_counts[param] ; i++)
1317
 
    {                                   /* Init multi locks */
1318
 
      multi_locks[i]= &data[i];
1319
 
      data[i].type= tests[param][i].lock_type;
1320
 
    }
1321
 
    thr_multi_lock(multi_locks, lock_counts[param], &owner);
1322
 
    pthread_mutex_lock(&LOCK_thread_count);
1323
 
    {
1324
 
      int tmp=rand() & 7;                       /* Do something from 0-2 sec */
1325
 
      if (tmp == 0)
1326
 
        sleep(1);
1327
 
      else if (tmp == 1)
1328
 
        sleep(2);
1329
 
      else
1330
 
      {
1331
 
        uint32_t k;
1332
 
        for (k=0 ; k < (uint32_t) (tmp-2)*100000L ; k++)
1333
 
          sum+=k;
1334
 
      }
1335
 
    }
1336
 
    pthread_mutex_unlock(&LOCK_thread_count);
1337
 
    thr_multi_unlock(multi_locks,lock_counts[param]);
1338
 
  }
1339
 
 
1340
 
  printf("Thread %s (%d) ended\n",my_thread_name(),param); fflush(stdout);
1341
 
  thr_print_locks();
1342
 
  pthread_mutex_lock(&LOCK_thread_count);
1343
 
  thread_count--;
1344
 
  pthread_cond_signal(&COND_thread_count); /* Tell main we are ready */
1345
 
  pthread_mutex_unlock(&LOCK_thread_count);
1346
 
  free((unsigned char*) arg);
1347
 
  return 0;
1348
 
}
1349
 
 
1350
 
 
1351
 
int main(int argc __attribute__((unused)),char **argv __attribute__((unused)))
1352
 
{
1353
 
  pthread_t tid;
1354
 
  pthread_attr_t thr_attr;
1355
 
  int i,*param,error;
1356
 
  MY_INIT(argv[0]);
1357
 
 
1358
 
  printf("Main thread: %s\n",my_thread_name());
1359
 
 
1360
 
  if ((error=pthread_cond_init(&COND_thread_count,NULL)))
1361
 
  {
1362
 
    fprintf(stderr,"Got error: %d from pthread_cond_init (errno: %d)",
1363
 
            error,errno);
1364
 
    exit(1);
1365
 
  }
1366
 
  if ((error=pthread_mutex_init(&LOCK_thread_count,MY_MUTEX_INIT_FAST)))
1367
 
  {
1368
 
    fprintf(stderr,"Got error: %d from pthread_cond_init (errno: %d)",
1369
 
            error,errno);
1370
 
    exit(1);
1371
 
  }
1372
 
 
1373
 
  for (i=0 ; i < (int) array_elements(locks) ; i++)
1374
 
  {
1375
 
    thr_lock_init(locks+i);
1376
 
    locks[i].check_status= test_check_status;
1377
 
    locks[i].update_status=test_update_status;
1378
 
    locks[i].copy_status=  test_copy_status;
1379
 
    locks[i].get_status=   test_get_status;
1380
 
  }
1381
 
  if ((error=pthread_attr_init(&thr_attr)))
1382
 
  {
1383
 
    fprintf(stderr,"Got error: %d from pthread_attr_init (errno: %d)",
1384
 
            error,errno);
1385
 
    exit(1);
1386
 
  }
1387
 
  if ((error=pthread_attr_setdetachstate(&thr_attr,PTHREAD_CREATE_DETACHED)))
1388
 
  {
1389
 
    fprintf(stderr,
1390
 
            "Got error: %d from pthread_attr_setdetachstate (errno: %d)",
1391
 
            error,errno);
1392
 
    exit(1);
1393
 
  }
1394
 
#ifndef pthread_attr_setstacksize               /* void return value */
1395
 
  if ((error=pthread_attr_setstacksize(&thr_attr,65536L)))
1396
 
  {
1397
 
    fprintf(stderr,"Got error: %d from pthread_attr_setstacksize (errno: %d)",
1398
 
            error,errno);
1399
 
    exit(1);
1400
 
  }
1401
 
#endif
1402
 
#ifdef HAVE_THR_SETCONCURRENCY
1403
 
  thr_setconcurrency(2);
1404
 
#endif
1405
 
  for (i=0 ; i < (int) array_elements(lock_counts) ; i++)
1406
 
  {
1407
 
    param=(int*) malloc(sizeof(int));
1408
 
    *param=i;
1409
 
 
1410
 
    if ((error=pthread_mutex_lock(&LOCK_thread_count)))
1411
 
    {
1412
 
      fprintf(stderr,"Got error: %d from pthread_mutex_lock (errno: %d)",
1413
 
              error,errno);
1414
 
      exit(1);
1415
 
    }
1416
 
    if ((error=pthread_create(&tid,&thr_attr,test_thread,(void*) param)))
1417
 
    {
1418
 
      fprintf(stderr,"Got error: %d from pthread_create (errno: %d)\n",
1419
 
              error,errno);
1420
 
      pthread_mutex_unlock(&LOCK_thread_count);
1421
 
      exit(1);
1422
 
    }
1423
 
    thread_count++;
1424
 
    pthread_mutex_unlock(&LOCK_thread_count);
1425
 
  }
1426
 
 
1427
 
  pthread_attr_destroy(&thr_attr);
1428
 
  if ((error=pthread_mutex_lock(&LOCK_thread_count)))
1429
 
    fprintf(stderr,"Got error: %d from pthread_mutex_lock\n",error);
1430
 
  while (thread_count)
1431
 
  {
1432
 
    if ((error=pthread_cond_wait(&COND_thread_count,&LOCK_thread_count)))
1433
 
      fprintf(stderr,"Got error: %d from pthread_cond_wait\n",error);
1434
 
  }
1435
 
  if ((error=pthread_mutex_unlock(&LOCK_thread_count)))
1436
 
    fprintf(stderr,"Got error: %d from pthread_mutex_unlock\n",error);
1437
 
  for (i=0 ; i < (int) array_elements(locks) ; i++)
1438
 
    thr_lock_delete(locks+i);
1439
 
#ifdef EXTRA_DEBUG
1440
 
  if (found_errors)
1441
 
    printf("Got %d warnings\n",found_errors);
1442
 
  else
1443
 
#endif
1444
 
    printf("Test succeeded\n");
1445
 
  return 0;
1446
 
}
1447
 
 
1448
 
#endif /* MAIN */