← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to mysys/thr_lock.cc

  • Committer: Jay Pipes
  • Date: 2009-02-21 16:00:06 UTC
  • mto: (907.1.1 trunk-with-temporal)
  • mto: This revision was merged to the branch mainline in revision 908.
  • Revision ID: jpipes@serialcoder-20090221160006-vnk3wt4qbcz62eru
Removes the TIME column type and related time functions.

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
29
30
TL_READ_NO_INSERT       # Read without concurrent inserts
30
31
TL_WRITE_ALLOW_WRITE    # Write lock that allows other writers
31
32
TL_WRITE_ALLOW_READ     # Write lock, but allow reading
32
33
TL_WRITE_CONCURRENT_INSERT
33
34
                        # 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
34
38
TL_WRITE                # High priority write
35
39
TL_WRITE_ONLY           # High priority write
36
40
                        # Abort all new lock request with an error
61
65
        for concurrent reads.
62
66
 
63
67
The lock algorithm allows one to have one TL_WRITE_ALLOW_READ,
64
 
TL_WRITE_CONCURRENT_INSERT lock at the same time as multiple read locks.
 
68
TL_WRITE_CONCURRENT_INSERT or one TL_WRITE_DELAYED lock at the same time as
 
69
multiple read locks.
65
70
 
66
71
*/
67
72
 
68
 
#include "config.h"
69
 
#include "drizzled/internal/my_sys.h"
 
73
#include "mysys_priv.h"
70
74
 
71
75
#include "thr_lock.h"
72
 
#include "drizzled/internal/m_string.h"
 
76
#include <mystrings/m_string.h>
73
77
#include <errno.h>
74
 
#include <list>
75
78
 
76
79
#if TIME_WITH_SYS_TIME
77
80
# include <sys/time.h>
86
89
 
87
90
#include <drizzled/util/test.h>
88
91
 
89
 
using namespace std;
90
 
 
91
 
namespace drizzled
92
 
{
93
 
 
94
 
bool thr_lock_inited= false;
 
92
bool thr_lock_inited=0;
95
93
uint32_t locks_immediate = 0L, locks_waited = 0L;
96
94
uint64_t table_lock_wait_timeout;
97
 
static enum thr_lock_type thr_upgraded_concurrent_insert_lock = TL_WRITE;
98
 
 
99
 
 
100
 
static list<THR_LOCK *> thr_lock_thread_list;          /* List of threads in use */
101
 
 
 
95
enum thr_lock_type thr_upgraded_concurrent_insert_lock = TL_WRITE;
 
96
 
 
97
 
 
98
LIST *thr_lock_thread_list;                     /* List of threads in use */
102
99
uint64_t max_write_lock_count= ~(uint64_t) 0L;
103
100
 
104
101
static inline pthread_cond_t *get_cond(void)
112
109
 
113
110
bool init_thr_lock()
114
111
{
115
 
  thr_lock_inited= true;
116
 
 
117
 
  return false;
 
112
  thr_lock_inited=1;
 
113
  return 0;
118
114
}
119
115
 
120
116
static inline bool
135
131
  lock->write_wait.last= &lock->write_wait.data;
136
132
  lock->write.last= &lock->write.data;
137
133
 
138
 
  pthread_mutex_lock(&internal::THR_LOCK_lock);         /* Add to locks in use */
139
 
  thr_lock_thread_list.push_front(lock);
140
 
  pthread_mutex_unlock(&internal::THR_LOCK_lock);
 
134
  pthread_mutex_lock(&THR_LOCK_lock);           /* Add to locks in use */
 
135
  lock->list.data=(void*) lock;
 
136
  thr_lock_thread_list=list_add(thr_lock_thread_list,&lock->list);
 
137
  pthread_mutex_unlock(&THR_LOCK_lock);
 
138
  return;
141
139
}
142
140
 
143
141
 
144
142
void thr_lock_delete(THR_LOCK *lock)
145
143
{
146
144
  pthread_mutex_destroy(&lock->mutex);
147
 
  pthread_mutex_lock(&internal::THR_LOCK_lock);
148
 
  thr_lock_thread_list.remove(lock);
149
 
  pthread_mutex_unlock(&internal::THR_LOCK_lock);
 
145
  pthread_mutex_lock(&THR_LOCK_lock);
 
146
  thr_lock_thread_list=list_delete(thr_lock_thread_list,&lock->list);
 
147
  pthread_mutex_unlock(&THR_LOCK_lock);
 
148
  return;
150
149
}
151
150
 
152
151
 
153
152
void thr_lock_info_init(THR_LOCK_INFO *info)
154
153
{
155
 
  internal::st_my_thread_var *tmp= my_thread_var;
156
 
  info->thread= tmp->pthread_self;
 
154
  struct st_my_thread_var *tmp= my_thread_var;
 
155
  info->thread=    tmp->pthread_self;
157
156
  info->thread_id= tmp->id;
158
157
  info->n_cursors= 0;
159
158
}
162
161
 
163
162
void thr_lock_data_init(THR_LOCK *lock,THR_LOCK_DATA *data, void *param)
164
163
{
165
 
  data->lock= lock;
166
 
  data->type= TL_UNLOCK;
167
 
  data->owner= NULL;                               /* no owner yet */
168
 
  data->status_param= param;
169
 
  data->cond= NULL;
 
164
  data->lock=lock;
 
165
  data->type=TL_UNLOCK;
 
166
  data->owner= 0;                               /* no owner yet */
 
167
  data->status_param=param;
 
168
  data->cond=0;
170
169
}
171
170
 
172
171
 
176
175
  for ( ; data ; data=data->next)
177
176
  {
178
177
    if (thr_lock_owner_equal(data->owner, owner))
179
 
      return true;                                      /* Already locked by thread */
180
 
  }
181
 
  return false;
182
 
}
 
178
      return 1;                                 /* Already locked by thread */
 
179
  }
 
180
  return 0;
 
181
}
 
182
 
 
183
static inline bool have_specific_lock(THR_LOCK_DATA *data,
 
184
                                         enum thr_lock_type type)
 
185
{
 
186
  for ( ; data ; data=data->next)
 
187
  {
 
188
    if (data->type == type)
 
189
      return 1;
 
190
  }
 
191
  return 0;
 
192
}
 
193
 
183
194
 
184
195
static void wake_up_waiters(THR_LOCK *lock);
185
196
 
188
199
wait_for_lock(struct st_lock_list *wait, THR_LOCK_DATA *data,
189
200
              bool in_wait_list)
190
201
{
191
 
  internal::st_my_thread_var *thread_var= my_thread_var;
 
202
  struct st_my_thread_var *thread_var= my_thread_var;
192
203
  pthread_cond_t *cond= &thread_var->suspend;
193
204
  struct timespec wait_timeout;
194
205
  enum enum_thr_lock_result result= THR_LOCK_ABORTED;
201
212
    wait->last= &data->next;
202
213
  }
203
214
 
204
 
  statistic_increment(locks_waited, &internal::THR_LOCK_lock);
 
215
  statistic_increment(locks_waited, &THR_LOCK_lock);
205
216
 
206
217
  /* Set up control struct to allow others to abort locks */
207
218
  thread_var->current_mutex= &data->lock->mutex;
235
246
    }
236
247
    if (rc == ETIMEDOUT || rc == ETIME)
237
248
    {
 
249
      /* purecov: begin inspected */
238
250
      result= THR_LOCK_WAIT_TIMEOUT;
239
251
      break;
 
252
      /* purecov: end */
240
253
    }
241
254
  }
242
255
  if (data->cond || data->type == TL_UNLOCK)
261
274
 
262
275
  /* The following must be done after unlock of lock->mutex */
263
276
  pthread_mutex_lock(&thread_var->mutex);
264
 
  thread_var->current_mutex= NULL;
265
 
  thread_var->current_cond= NULL;
 
277
  thread_var->current_mutex= 0;
 
278
  thread_var->current_cond=  0;
266
279
  pthread_mutex_unlock(&thread_var->mutex);
267
280
  return(result);
268
281
}
269
282
 
270
283
 
271
 
static enum enum_thr_lock_result
 
284
enum enum_thr_lock_result
272
285
thr_lock(THR_LOCK_DATA *data, THR_LOCK_OWNER *owner,
273
286
         enum thr_lock_type lock_type)
274
287
{
297
310
      */
298
311
 
299
312
      if (thr_lock_owner_equal(data->owner, lock->write.data->owner) ||
300
 
          (lock->write.data->type <= TL_WRITE_CONCURRENT_INSERT &&
301
 
           (((int) lock_type <= (int) TL_READ_WITH_SHARED_LOCKS) ||
 
313
          (lock->write.data->type <= TL_WRITE_DELAYED &&
 
314
           (((int) lock_type <= (int) TL_READ_HIGH_PRIORITY) ||
302
315
            (lock->write.data->type != TL_WRITE_CONCURRENT_INSERT &&
303
316
             lock->write.data->type != TL_WRITE_ALLOW_READ))))
304
317
      {                                         /* Already got a write lock */
309
322
          lock->read_no_write_count++;
310
323
        if (lock->get_status)
311
324
          (*lock->get_status)(data->status_param, 0);
312
 
        statistic_increment(locks_immediate,&internal::THR_LOCK_lock);
 
325
        statistic_increment(locks_immediate,&THR_LOCK_lock);
313
326
        goto end;
314
327
      }
315
328
      if (lock->write.data->type == TL_WRITE_ONLY)
321
334
      }
322
335
    }
323
336
    else if (!lock->write_wait.data ||
324
 
             lock->write_wait.data->type <= TL_WRITE_DEFAULT ||
 
337
             lock->write_wait.data->type <= TL_WRITE_LOW_PRIORITY ||
 
338
             lock_type == TL_READ_HIGH_PRIORITY ||
325
339
             have_old_read_lock(lock->read.data, data->owner))
326
340
    {                                           /* No important write-locks */
327
341
      (*lock->read.last)=data;                  /* Add to running FIFO */
331
345
        (*lock->get_status)(data->status_param, 0);
332
346
      if (lock_type == TL_READ_NO_INSERT)
333
347
        lock->read_no_write_count++;
334
 
      statistic_increment(locks_immediate,&internal::THR_LOCK_lock);
 
348
      statistic_increment(locks_immediate,&THR_LOCK_lock);
335
349
      goto end;
336
350
    }
337
351
    /*
343
357
  }
344
358
  else                                          /* Request for WRITE lock */
345
359
  {
346
 
    if (lock_type == TL_WRITE_CONCURRENT_INSERT && ! lock->check_status)
 
360
    if (lock_type == TL_WRITE_DELAYED)
 
361
    {
 
362
      if (lock->write.data && lock->write.data->type == TL_WRITE_ONLY)
 
363
      {
 
364
        data->type=TL_UNLOCK;
 
365
        result= THR_LOCK_ABORTED;               /* Can't wait for this one */
 
366
        goto end;
 
367
      }
 
368
      /*
 
369
        if there is a TL_WRITE_ALLOW_READ lock, we have to wait for a lock
 
370
        (TL_WRITE_ALLOW_READ is used for ALTER TABLE in MySQL)
 
371
      */
 
372
      if ((!lock->write.data ||
 
373
           lock->write.data->type != TL_WRITE_ALLOW_READ) &&
 
374
          !have_specific_lock(lock->write_wait.data,TL_WRITE_ALLOW_READ) &&
 
375
          (lock->write.data || lock->read.data))
 
376
      {
 
377
        /* Add delayed write lock to write_wait queue, and return at once */
 
378
        (*lock->write_wait.last)=data;
 
379
        data->prev=lock->write_wait.last;
 
380
        lock->write_wait.last= &data->next;
 
381
        data->cond=get_cond();
 
382
        /*
 
383
          We don't have to do get_status here as we will do it when we change
 
384
          the delayed lock to a real write lock
 
385
        */
 
386
        statistic_increment(locks_immediate,&THR_LOCK_lock);
 
387
        goto end;
 
388
      }
 
389
    }
 
390
    else if (lock_type == TL_WRITE_CONCURRENT_INSERT && ! lock->check_status)
347
391
      data->type=lock_type= thr_upgraded_concurrent_insert_lock;
348
392
 
349
393
    if (lock->write.data)                       /* If there is a write lock */
380
424
        lock->write.last= &data->next;
381
425
        if (data->lock->get_status)
382
426
          (*data->lock->get_status)(data->status_param, 0);
383
 
        statistic_increment(locks_immediate,&internal::THR_LOCK_lock);
 
427
        statistic_increment(locks_immediate,&THR_LOCK_lock);
384
428
        goto end;
385
429
      }
386
430
    }
400
444
        }
401
445
 
402
446
        if (!lock->read.data ||
403
 
            (lock_type <= TL_WRITE_CONCURRENT_INSERT &&
 
447
            (lock_type <= TL_WRITE_DELAYED &&
404
448
             ((lock_type != TL_WRITE_CONCURRENT_INSERT &&
405
449
               lock_type != TL_WRITE_ALLOW_WRITE) ||
406
450
              !lock->read_no_write_count)))
410
454
          lock->write.last= &data->next;
411
455
          if (data->lock->get_status)
412
456
            (*data->lock->get_status)(data->status_param, concurrent_insert);
413
 
          statistic_increment(locks_immediate,&internal::THR_LOCK_lock);
 
457
          statistic_increment(locks_immediate,&THR_LOCK_lock);
414
458
          goto end;
415
459
        }
416
460
      }
437
481
}
438
482
 
439
483
 
440
 
static void free_all_read_locks(THR_LOCK *lock, bool using_concurrent_insert)
 
484
static inline void free_all_read_locks(THR_LOCK *lock,
 
485
                                       bool using_concurrent_insert)
441
486
{
442
487
  THR_LOCK_DATA *data=lock->read_wait.data;
443
488
 
481
526
 
482
527
        /* Unlock lock and free next thread on same lock */
483
528
 
484
 
static void thr_unlock(THR_LOCK_DATA *data)
 
529
void thr_unlock(THR_LOCK_DATA *data)
485
530
{
486
531
  THR_LOCK *lock=data->lock;
487
532
  enum thr_lock_type lock_type=data->type;
491
536
    data->next->prev= data->prev;
492
537
  else if (lock_type <= TL_READ_NO_INSERT)
493
538
    lock->read.last=data->prev;
 
539
  else if (lock_type == TL_WRITE_DELAYED && data->cond)
 
540
  {
 
541
    /*
 
542
      This only happens in extreme circumstances when a
 
543
      write delayed lock that is waiting for a lock
 
544
    */
 
545
    lock->write_wait.last=data->prev;           /* Put it on wait queue */
 
546
  }
494
547
  else
495
548
    lock->write.last=data->prev;
496
549
  if (lock_type >= TL_WRITE_CONCURRENT_INSERT)
532
585
    {
533
586
      /* Release write-locks with TL_WRITE or TL_WRITE_ONLY priority first */
534
587
      if (data &&
535
 
          (!lock->read_wait.data || lock->read_wait.data->type <= TL_READ_WITH_SHARED_LOCKS))
 
588
          (data->type != TL_WRITE_LOW_PRIORITY || !lock->read_wait.data ||
 
589
           lock->read_wait.data->type < TL_READ_HIGH_PRIORITY))
536
590
      {
537
591
        if (lock->write_lock_count++ > max_write_lock_count)
538
592
        {
568
622
            break;
569
623
          data=lock->write_wait.data;           /* Free this too */
570
624
        }
571
 
        if (data->type >= TL_WRITE)
 
625
        if (data->type >= TL_WRITE_LOW_PRIORITY)
572
626
          goto end;
573
627
        /* Release possible read locks together with the write lock */
574
628
      }
579
633
                             data->type == TL_WRITE_ALLOW_WRITE));
580
634
    }
581
635
    else if (data &&
582
 
             (lock_type=data->type) <= TL_WRITE_CONCURRENT_INSERT &&
 
636
             (lock_type=data->type) <= TL_WRITE_DELAYED &&
583
637
             ((lock_type != TL_WRITE_CONCURRENT_INSERT &&
584
638
               lock_type != TL_WRITE_ALLOW_WRITE) ||
585
639
              !lock->read_no_write_count))
739
793
  TL_WRITE_ONLY to abort any new accesses to the lock
740
794
*/
741
795
 
742
 
void thr_abort_locks(THR_LOCK *lock)
 
796
void thr_abort_locks(THR_LOCK *lock, bool upgrade_lock)
743
797
{
744
798
  THR_LOCK_DATA *data;
745
799
  pthread_mutex_lock(&lock->mutex);
746
800
 
747
801
  for (data=lock->read_wait.data; data ; data=data->next)
748
802
  {
749
 
    data->type= TL_UNLOCK;                      /* Mark killed */
 
803
    data->type=TL_UNLOCK;                       /* Mark killed */
750
804
    /* It's safe to signal the cond first: we're still holding the mutex. */
751
805
    pthread_cond_signal(data->cond);
752
 
    data->cond= NULL;                           /* Removed from list */
 
806
    data->cond=0;                               /* Removed from list */
753
807
  }
754
808
  for (data=lock->write_wait.data; data ; data=data->next)
755
809
  {
756
810
    data->type=TL_UNLOCK;
757
811
    pthread_cond_signal(data->cond);
758
 
    data->cond= NULL;
 
812
    data->cond=0;
759
813
  }
760
814
  lock->read_wait.last= &lock->read_wait.data;
761
815
  lock->write_wait.last= &lock->write_wait.data;
762
816
  lock->read_wait.data=lock->write_wait.data=0;
763
 
  if (lock->write.data)
 
817
  if (upgrade_lock && lock->write.data)
764
818
    lock->write.data->type=TL_WRITE_ONLY;
765
819
  pthread_mutex_unlock(&lock->mutex);
766
820
  return;
773
827
  This is used to abort all locks for a specific thread
774
828
*/
775
829
 
776
 
bool thr_abort_locks_for_thread(THR_LOCK *lock, uint64_t thread_id)
 
830
bool thr_abort_locks_for_thread(THR_LOCK *lock, my_thread_id thread_id)
777
831
{
778
832
  THR_LOCK_DATA *data;
779
833
  bool found= false;
781
835
  pthread_mutex_lock(&lock->mutex);
782
836
  for (data= lock->read_wait.data; data ; data= data->next)
783
837
  {
784
 
    if (data->owner->info->thread_id == thread_id)
 
838
    if (data->owner->info->thread_id == thread_id)    /* purecov: tested */
785
839
    {
786
840
      data->type= TL_UNLOCK;                    /* Mark killed */
787
841
      /* It's safe to signal the cond first: we're still holding the mutex. */
797
851
  }
798
852
  for (data= lock->write_wait.data; data ; data= data->next)
799
853
  {
800
 
    if (data->owner->info->thread_id == thread_id)
 
854
    if (data->owner->info->thread_id == thread_id) /* purecov: tested */
801
855
    {
802
856
      data->type= TL_UNLOCK;
803
857
      found= true;
804
858
      pthread_cond_signal(data->cond);
805
 
      data->cond= NULL;
 
859
      data->cond= 0;
806
860
 
807
861
      if (((*data->prev)= data->next))
808
862
        data->next->prev= data->prev;
815
869
  return(found);
816
870
}
817
871
 
818
 
} /* namespace drizzled */
 
872
 
 
873
/*
 
874
  Downgrade a WRITE_* to a lower WRITE level
 
875
  SYNOPSIS
 
876
    thr_downgrade_write_lock()
 
877
    in_data                   Lock data of thread downgrading its lock
 
878
    new_lock_type             New write lock type
 
879
  RETURN VALUE
 
880
    NONE
 
881
  DESCRIPTION
 
882
    This can be used to downgrade a lock already owned. When the downgrade
 
883
    occurs also other waiters, both readers and writers can be allowed to
 
884
    start.
 
885
    The previous lock is often TL_WRITE_ONLY but can also be
 
886
    TL_WRITE and TL_WRITE_ALLOW_READ. The normal downgrade variants are
 
887
    TL_WRITE_ONLY => TL_WRITE_ALLOW_READ After a short exclusive lock
 
888
    TL_WRITE_ALLOW_READ => TL_WRITE_ALLOW_WRITE After discovering that the
 
889
    operation didn't need such a high lock.
 
890
    TL_WRITE_ONLY => TL_WRITE after a short exclusive lock while holding a
 
891
    write table lock
 
892
    TL_WRITE_ONLY => TL_WRITE_ALLOW_WRITE After a short exclusive lock after
 
893
    already earlier having dongraded lock to TL_WRITE_ALLOW_WRITE
 
894
    The implementation is conservative and rather don't start rather than
 
895
    go on unknown paths to start, the common cases are handled.
 
896
 
 
897
    NOTE:
 
898
    In its current implementation it is only allowed to downgrade from
 
899
    TL_WRITE_ONLY. In this case there are no waiters. Thus no wake up
 
900
    logic is required.
 
901
*/
 
902
 
 
903
void thr_downgrade_write_lock(THR_LOCK_DATA *in_data,
 
904
                              enum thr_lock_type new_lock_type)
 
905
{
 
906
  THR_LOCK *lock=in_data->lock;
 
907
 
 
908
  pthread_mutex_lock(&lock->mutex);
 
909
  in_data->type= new_lock_type;
 
910
 
 
911
  pthread_mutex_unlock(&lock->mutex);
 
912
  return;
 
913
}
 
914
 
 
915
/* Upgrade a WRITE_DELAY lock to a WRITE_LOCK */
 
916
 
 
917
bool thr_upgrade_write_delay_lock(THR_LOCK_DATA *data)
 
918
{
 
919
  THR_LOCK *lock=data->lock;
 
920
 
 
921
  pthread_mutex_lock(&lock->mutex);
 
922
  if (data->type == TL_UNLOCK || data->type >= TL_WRITE_LOW_PRIORITY)
 
923
  {
 
924
    pthread_mutex_unlock(&lock->mutex);
 
925
    return(data->type == TL_UNLOCK);    /* Test if Aborted */
 
926
  }
 
927
  /* TODO:  Upgrade to TL_WRITE_CONCURRENT_INSERT in some cases */
 
928
  data->type=TL_WRITE;                          /* Upgrade lock */
 
929
 
 
930
  /* Check if someone has given us the lock */
 
931
  if (!data->cond)
 
932
  {
 
933
    if (!lock->read.data)                       /* No read locks */
 
934
    {                                           /* We have the lock */
 
935
      if (data->lock->get_status)
 
936
        (*data->lock->get_status)(data->status_param, 0);
 
937
      pthread_mutex_unlock(&lock->mutex);
 
938
      return(0);
 
939
    }
 
940
 
 
941
    if (((*data->prev)=data->next))             /* remove from lock-list */
 
942
      data->next->prev= data->prev;
 
943
    else
 
944
      lock->write.last=data->prev;
 
945
 
 
946
    if ((data->next=lock->write_wait.data))     /* Put first in lock_list */
 
947
      data->next->prev= &data->next;
 
948
    else
 
949
      lock->write_wait.last= &data->next;
 
950
    data->prev= &lock->write_wait.data;
 
951
    lock->write_wait.data=data;
 
952
  }
 
953
 
 
954
  return(wait_for_lock(&lock->write_wait,data,1));
 
955
}
 
956
 
 
957
 
 
958
/* downgrade a WRITE lock to a WRITE_DELAY lock if there is pending locks */
 
959
 
 
960
bool thr_reschedule_write_lock(THR_LOCK_DATA *data)
 
961
{
 
962
  THR_LOCK *lock=data->lock;
 
963
 
 
964
  pthread_mutex_lock(&lock->mutex);
 
965
  if (!lock->read_wait.data)                    /* No waiting read locks */
 
966
  {
 
967
    pthread_mutex_unlock(&lock->mutex);
 
968
    return(0);
 
969
  }
 
970
 
 
971
  data->type=TL_WRITE_DELAYED;
 
972
  if (lock->update_status)
 
973
    (*lock->update_status)(data->status_param);
 
974
  if (((*data->prev)=data->next))               /* remove from lock-list */
 
975
    data->next->prev= data->prev;
 
976
  else
 
977
    lock->write.last=data->prev;
 
978
 
 
979
  if ((data->next=lock->write_wait.data))       /* Put first in lock_list */
 
980
    data->next->prev= &data->next;
 
981
  else
 
982
    lock->write_wait.last= &data->next;
 
983
  data->prev= &lock->write_wait.data;
 
984
  data->cond=get_cond();                        /* This was zero */
 
985
  lock->write_wait.data=data;
 
986
  free_all_read_locks(lock,0);
 
987
 
 
988
  pthread_mutex_unlock(&lock->mutex);
 
989
  return(thr_upgrade_write_delay_lock(data));
 
990
}