← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to mysys/thr_lock.cc

  • Committer: Stewart Smith
  • Date: 2009-03-11 06:37:19 UTC
  • mto: (910.4.19 sparc) (937.2.1 sparc)
  • mto: This revision was merged to the branch mainline in revision 931.
  • Revision ID: stewart@flamingspork.com-20090311063719-v9iqjd00ts6260vv
batch up more INSERTs into transactions to help tests run quicker.

Show diffs side-by-side

added added

removed removed

Lines of Context:
mysys/thr_lock.cc
11
11
 
12
12
   You should have received a copy of the GNU General Public License
13
13
   along with this program; if not, write to the Free Software
14
 
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA */
 
14
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
15
15
 
16
16
/*
17
17
Read and write locks for Posix threads. All tread must acquire
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
46
50
should put a pointer to the following functions in the lock structure:
47
51
(If the pointer is zero (default), the function is not called)
48
52
 
 
53
check_status:
 
54
         Before giving a lock of type TL_WRITE_CONCURRENT_INSERT,
 
55
         we check if this function exists and returns 0.
 
56
         If not, then the lock is upgraded to TL_WRITE_LOCK
 
57
         In MyISAM this is a simple check if the insert can be done
 
58
         at the end of the datafile.
 
59
update_status:
 
60
        Before a write lock is released, this function is called.
 
61
        In MyISAM this functions updates the count and length of the datafile
 
62
get_status:
 
63
        When one gets a lock this functions is called.
 
64
        In MyISAM this stores the number of rows and size of the datafile
 
65
        for concurrent reads.
49
66
 
50
67
The lock algorithm allows one to have one TL_WRITE_ALLOW_READ,
51
 
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.
52
70
 
53
71
*/
54
72
 
55
 
#include "config.h"
56
 
#include "drizzled/internal/my_sys.h"
57
 
#include "drizzled/internal/thread_var.h"
58
 
#include "drizzled/statistics_variables.h"
59
 
#include "drizzled/pthread_globals.h"
60
 
 
61
 
#include "drizzled/session.h"
 
73
#include "mysys_priv.h"
62
74
 
63
75
#include "thr_lock.h"
64
 
#include "drizzled/internal/m_string.h"
 
76
#include <mystrings/m_string.h>
65
77
#include <errno.h>
66
 
#include <list>
67
78
 
68
79
#if TIME_WITH_SYS_TIME
69
80
# include <sys/time.h>
78
89
 
79
90
#include <drizzled/util/test.h>
80
91
 
81
 
#include <boost/interprocess/sync/lock_options.hpp>
82
 
 
83
 
using namespace std;
84
 
 
85
 
namespace drizzled
86
 
{
87
 
 
 
92
bool thr_lock_inited=0;
 
93
uint32_t locks_immediate = 0L, locks_waited = 0L;
88
94
uint64_t table_lock_wait_timeout;
89
 
static enum thr_lock_type thr_upgraded_concurrent_insert_lock = TL_WRITE;
90
 
 
91
 
 
92
 
uint64_t max_write_lock_count= UINT64_MAX;
93
 
 
94
 
void thr_multi_unlock(THR_LOCK_DATA **data,uint32_t count);
 
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 */
 
99
uint64_t max_write_lock_count= ~(uint64_t) 0L;
 
100
 
 
101
static inline pthread_cond_t *get_cond(void)
 
102
{
 
103
  return &my_thread_var->suspend;
 
104
}
95
105
 
96
106
/*
97
107
** For the future (now the thread specific cond is alloced by my_pthread.c)
98
108
*/
99
109
 
 
110
bool init_thr_lock()
 
111
{
 
112
  thr_lock_inited=1;
 
113
  return 0;
 
114
}
 
115
 
100
116
static inline bool
101
117
thr_lock_owner_equal(THR_LOCK_OWNER *rhs, THR_LOCK_OWNER *lhs)
102
118
{
108
124
 
109
125
void thr_lock_init(THR_LOCK *lock)
110
126
{
111
 
  lock->init();
 
127
  memset(lock, 0, sizeof(*lock));
 
128
  pthread_mutex_init(&lock->mutex,MY_MUTEX_INIT_FAST);
112
129
  lock->read.last= &lock->read.data;
113
130
  lock->read_wait.last= &lock->read_wait.data;
114
131
  lock->write_wait.last= &lock->write_wait.data;
115
132
  lock->write.last= &lock->write.data;
116
 
}
117
 
 
118
 
 
119
 
void THR_LOCK_INFO::init()
120
 
{
121
 
  internal::st_my_thread_var *tmp= my_thread_var;
122
 
  thread_id= tmp->id;
123
 
  n_cursors= 0;
 
133
 
 
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;
 
139
}
 
140
 
 
141
 
 
142
void thr_lock_delete(THR_LOCK *lock)
 
143
{
 
144
  pthread_mutex_destroy(&lock->mutex);
 
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;
 
149
}
 
150
 
 
151
 
 
152
void thr_lock_info_init(THR_LOCK_INFO *info)
 
153
{
 
154
  struct st_my_thread_var *tmp= my_thread_var;
 
155
  info->thread=    tmp->pthread_self;
 
156
  info->thread_id= tmp->id;
 
157
  info->n_cursors= 0;
124
158
}
125
159
 
126
160
        /* Initialize a lock instance */
127
161
 
128
 
void THR_LOCK_DATA::init(THR_LOCK *lock_arg, void *param_arg)
 
162
void thr_lock_data_init(THR_LOCK *lock,THR_LOCK_DATA *data, void *param)
129
163
{
130
 
  lock= lock_arg;
131
 
  type= TL_UNLOCK;
132
 
  owner= NULL;                               /* no owner yet */
133
 
  status_param= param_arg;
134
 
  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;
135
169
}
136
170
 
137
171
 
141
175
  for ( ; data ; data=data->next)
142
176
  {
143
177
    if (thr_lock_owner_equal(data->owner, owner))
144
 
      return true;                                      /* Already locked by thread */
145
 
  }
146
 
  return false;
147
 
}
 
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
 
148
194
 
149
195
static void wake_up_waiters(THR_LOCK *lock);
150
196
 
151
197
 
152
 
static enum enum_thr_lock_result wait_for_lock(Session &session, struct st_lock_list *wait, THR_LOCK_DATA *data)
 
198
static enum enum_thr_lock_result
 
199
wait_for_lock(struct st_lock_list *wait, THR_LOCK_DATA *data,
 
200
              bool in_wait_list)
153
201
{
154
 
  internal::st_my_thread_var *thread_var= session.getThreadVar();
155
 
 
156
 
  boost::condition_variable_any *cond= &thread_var->suspend;
 
202
  struct st_my_thread_var *thread_var= my_thread_var;
 
203
  pthread_cond_t *cond= &thread_var->suspend;
 
204
  struct timespec wait_timeout;
157
205
  enum enum_thr_lock_result result= THR_LOCK_ABORTED;
158
206
  bool can_deadlock= test(data->owner->info->n_cursors);
159
207
 
 
208
  if (!in_wait_list)
160
209
  {
161
210
    (*wait->last)=data;                         /* Wait for lock */
162
211
    data->prev= wait->last;
163
212
    wait->last= &data->next;
164
213
  }
165
214
 
166
 
  current_global_counters.locks_waited++;
 
215
  statistic_increment(locks_waited, &THR_LOCK_lock);
167
216
 
168
217
  /* Set up control struct to allow others to abort locks */
169
 
  thread_var->current_mutex= data->lock->native_handle();
170
 
  thread_var->current_cond=  &thread_var->suspend;
171
 
  data->cond= &thread_var->suspend;;
 
218
  thread_var->current_mutex= &data->lock->mutex;
 
219
  thread_var->current_cond=  cond;
 
220
  data->cond= cond;
172
221
 
173
 
  while (not thread_var->abort)
 
222
  if (can_deadlock)
 
223
    set_timespec(wait_timeout, table_lock_wait_timeout);
 
224
  while (!thread_var->abort || in_wait_list)
174
225
  {
175
 
    boost_unique_lock_t scoped(*data->lock->native_handle(), boost::adopt_lock_t());
176
 
 
177
 
    if (can_deadlock)
178
 
    {
179
 
      boost::xtime xt; 
180
 
      xtime_get(&xt, boost::TIME_UTC); 
181
 
      xt.sec += table_lock_wait_timeout; 
182
 
      if (not cond->timed_wait(scoped, xt))
183
 
      {
184
 
        result= THR_LOCK_WAIT_TIMEOUT;
185
 
        scoped.release();
186
 
        break;
187
 
      }
188
 
    }
189
 
    else
190
 
    {
191
 
      cond->wait(scoped);
192
 
    }
 
226
    int rc= (can_deadlock ?
 
227
             pthread_cond_timedwait(cond, &data->lock->mutex,
 
228
                                    &wait_timeout) :
 
229
             pthread_cond_wait(cond, &data->lock->mutex));
193
230
    /*
194
231
      We must break the wait if one of the following occurs:
195
232
      - the connection has been aborted (!thread_var->abort), but
203
240
      Order of checks below is important to not report about timeout
204
241
      if the predicate is true.
205
242
    */
206
 
    if (data->cond == NULL)
207
 
    {
208
 
      scoped.release();
209
 
      break;
210
 
    }
211
 
    scoped.release();
 
243
    if (data->cond == 0)
 
244
    {
 
245
      break;
 
246
    }
 
247
    if (rc == ETIMEDOUT || rc == ETIME)
 
248
    {
 
249
      /* purecov: begin inspected */
 
250
      result= THR_LOCK_WAIT_TIMEOUT;
 
251
      break;
 
252
      /* purecov: end */
 
253
    }
212
254
  }
213
255
  if (data->cond || data->type == TL_UNLOCK)
214
256
  {
225
267
  else
226
268
  {
227
269
    result= THR_LOCK_SUCCESS;
 
270
    if (data->lock->get_status)
 
271
      (*data->lock->get_status)(data->status_param, 0);
228
272
  }
229
 
  data->lock->unlock();
 
273
  pthread_mutex_unlock(&data->lock->mutex);
230
274
 
231
275
  /* The following must be done after unlock of lock->mutex */
232
 
  boost_unique_lock_t scopedLock(thread_var->mutex);
233
 
  thread_var->current_mutex= NULL;
234
 
  thread_var->current_cond= NULL;
 
276
  pthread_mutex_lock(&thread_var->mutex);
 
277
  thread_var->current_mutex= 0;
 
278
  thread_var->current_cond=  0;
 
279
  pthread_mutex_unlock(&thread_var->mutex);
235
280
  return(result);
236
281
}
237
282
 
238
283
 
239
 
static enum enum_thr_lock_result thr_lock(Session &session, THR_LOCK_DATA *data, THR_LOCK_OWNER *owner, enum thr_lock_type lock_type)
 
284
enum enum_thr_lock_result
 
285
thr_lock(THR_LOCK_DATA *data, THR_LOCK_OWNER *owner,
 
286
         enum thr_lock_type lock_type)
240
287
{
241
 
  THR_LOCK *lock= data->lock;
 
288
  THR_LOCK *lock=data->lock;
242
289
  enum enum_thr_lock_result result= THR_LOCK_SUCCESS;
243
290
  struct st_lock_list *wait_queue;
244
291
  THR_LOCK_DATA *lock_owner;
247
294
  data->cond=0;                                 /* safety */
248
295
  data->type=lock_type;
249
296
  data->owner= owner;                           /* Must be reset ! */
250
 
  lock->lock();
 
297
  pthread_mutex_lock(&lock->mutex);
251
298
  if ((int) lock_type <= (int) TL_READ_NO_INSERT)
252
299
  {
253
300
    /* Request for READ lock */
263
310
      */
264
311
 
265
312
      if (thr_lock_owner_equal(data->owner, lock->write.data->owner) ||
266
 
          (lock->write.data->type <= TL_WRITE_CONCURRENT_INSERT &&
267
 
           (((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) ||
268
315
            (lock->write.data->type != TL_WRITE_CONCURRENT_INSERT &&
269
316
             lock->write.data->type != TL_WRITE_ALLOW_READ))))
270
317
      {                                         /* Already got a write lock */
273
320
        lock->read.last= &data->next;
274
321
        if (lock_type == TL_READ_NO_INSERT)
275
322
          lock->read_no_write_count++;
276
 
        current_global_counters.locks_immediate++;
 
323
        if (lock->get_status)
 
324
          (*lock->get_status)(data->status_param, 0);
 
325
        statistic_increment(locks_immediate,&THR_LOCK_lock);
277
326
        goto end;
278
327
      }
279
328
      if (lock->write.data->type == TL_WRITE_ONLY)
285
334
      }
286
335
    }
287
336
    else if (!lock->write_wait.data ||
288
 
             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 ||
289
339
             have_old_read_lock(lock->read.data, data->owner))
290
340
    {                                           /* No important write-locks */
291
341
      (*lock->read.last)=data;                  /* Add to running FIFO */
292
342
      data->prev=lock->read.last;
293
343
      lock->read.last= &data->next;
 
344
      if (lock->get_status)
 
345
        (*lock->get_status)(data->status_param, 0);
294
346
      if (lock_type == TL_READ_NO_INSERT)
295
347
        lock->read_no_write_count++;
296
 
      current_global_counters.locks_immediate++;
 
348
      statistic_increment(locks_immediate,&THR_LOCK_lock);
297
349
      goto end;
298
350
    }
299
351
    /*
305
357
  }
306
358
  else                                          /* Request for WRITE lock */
307
359
  {
308
 
    if (lock_type == TL_WRITE_CONCURRENT_INSERT)
 
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)
309
391
      data->type=lock_type= thr_upgraded_concurrent_insert_lock;
310
392
 
311
393
    if (lock->write.data)                       /* If there is a write lock */
340
422
        (*lock->write.last)=data;       /* Add to running fifo */
341
423
        data->prev=lock->write.last;
342
424
        lock->write.last= &data->next;
343
 
        current_global_counters.locks_immediate++;
 
425
        if (data->lock->get_status)
 
426
          (*data->lock->get_status)(data->status_param, 0);
 
427
        statistic_increment(locks_immediate,&THR_LOCK_lock);
344
428
        goto end;
345
429
      }
346
430
    }
352
436
        if (lock_type == TL_WRITE_CONCURRENT_INSERT)
353
437
        {
354
438
          concurrent_insert= 1;
 
439
          if ((*lock->check_status)(data->status_param))
 
440
          {
 
441
            concurrent_insert= 0;
 
442
            data->type=lock_type= thr_upgraded_concurrent_insert_lock;
 
443
          }
355
444
        }
356
445
 
357
446
        if (!lock->read.data ||
358
 
            (lock_type <= TL_WRITE_CONCURRENT_INSERT &&
 
447
            (lock_type <= TL_WRITE_DELAYED &&
359
448
             ((lock_type != TL_WRITE_CONCURRENT_INSERT &&
360
449
               lock_type != TL_WRITE_ALLOW_WRITE) ||
361
450
              !lock->read_no_write_count)))
363
452
          (*lock->write.last)=data;             /* Add as current write lock */
364
453
          data->prev=lock->write.last;
365
454
          lock->write.last= &data->next;
366
 
          current_global_counters.locks_immediate++;
 
455
          if (data->lock->get_status)
 
456
            (*data->lock->get_status)(data->status_param, concurrent_insert);
 
457
          statistic_increment(locks_immediate,&THR_LOCK_lock);
367
458
          goto end;
368
459
        }
369
460
      }
382
473
    result= THR_LOCK_DEADLOCK;
383
474
    goto end;
384
475
  }
385
 
 
386
476
  /* Can't get lock yet;  Wait for it */
387
 
  return(wait_for_lock(session, wait_queue, data));
 
477
  return(wait_for_lock(wait_queue, data, 0));
388
478
end:
389
 
  lock->unlock();
390
 
 
 
479
  pthread_mutex_unlock(&lock->mutex);
391
480
  return(result);
392
481
}
393
482
 
394
483
 
395
 
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)
396
486
{
397
 
  THR_LOCK_DATA *data= lock->read_wait.data;
 
487
  THR_LOCK_DATA *data=lock->read_wait.data;
398
488
 
399
489
  /* move all locks from read_wait list to read list */
400
490
  (*lock->read.last)=data;
406
496
 
407
497
  do
408
498
  {
409
 
    boost::condition_variable_any *cond= data->cond;
 
499
    pthread_cond_t *cond=data->cond;
410
500
    if ((int) data->type == (int) TL_READ_NO_INSERT)
411
501
    {
412
502
      if (using_concurrent_insert)
426
516
      }
427
517
      lock->read_no_write_count++;
428
518
    }
429
 
    data->cond= NULL;                           /* Mark thread free */
430
 
    cond->notify_one();
 
519
    data->cond=0;                               /* Mark thread free */
 
520
    pthread_cond_signal(cond);
431
521
  } while ((data=data->next));
432
522
  *lock->read_wait.last=0;
433
523
  if (!lock->read_wait.data)
434
524
    lock->write_lock_count=0;
435
525
}
436
526
 
437
 
/* Unlock lock and free next thread on same lock */
 
527
        /* Unlock lock and free next thread on same lock */
438
528
 
439
 
static void thr_unlock(THR_LOCK_DATA *data)
 
529
void thr_unlock(THR_LOCK_DATA *data)
440
530
{
441
531
  THR_LOCK *lock=data->lock;
442
532
  enum thr_lock_type lock_type=data->type;
443
 
  lock->lock();
 
533
  pthread_mutex_lock(&lock->mutex);
444
534
 
445
535
  if (((*data->prev)=data->next))               /* remove from lock-list */
446
536
    data->next->prev= data->prev;
447
537
  else if (lock_type <= TL_READ_NO_INSERT)
448
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
  }
449
547
  else
450
548
    lock->write.last=data->prev;
451
549
  if (lock_type >= TL_WRITE_CONCURRENT_INSERT)
452
 
  { }
 
550
  {
 
551
    if (lock->update_status)
 
552
      (*lock->update_status)(data->status_param);
 
553
  }
453
554
  else
454
 
  { }
 
555
  {
 
556
    if (lock->restore_status)
 
557
      (*lock->restore_status)(data->status_param);
 
558
  }
455
559
  if (lock_type == TL_READ_NO_INSERT)
456
560
    lock->read_no_write_count--;
457
561
  data->type=TL_UNLOCK;                         /* Mark unlocked */
458
562
  wake_up_waiters(lock);
459
 
  lock->unlock();
 
563
  pthread_mutex_unlock(&lock->mutex);
 
564
  return;
460
565
}
461
566
 
462
567
 
480
585
    {
481
586
      /* Release write-locks with TL_WRITE or TL_WRITE_ONLY priority first */
482
587
      if (data &&
483
 
          (!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))
484
590
      {
485
591
        if (lock->write_lock_count++ > max_write_lock_count)
486
592
        {
502
608
          data->prev=lock->write.last;
503
609
          data->next=0;
504
610
          lock->write.last= &data->next;
505
 
 
 
611
          if (data->type == TL_WRITE_CONCURRENT_INSERT &&
 
612
              (*lock->check_status)(data->status_param))
 
613
            data->type=TL_WRITE;                        /* Upgrade lock */
506
614
          {
507
 
            boost::condition_variable_any *cond= data->cond;
508
 
            data->cond= NULL;                           /* Mark thread free */
509
 
            cond->notify_one(); /* Start waiting thred */
 
615
            pthread_cond_t *cond=data->cond;
 
616
            data->cond=0;                               /* Mark thread free */
 
617
            pthread_cond_signal(cond);  /* Start waiting thread */
510
618
          }
511
619
          if (data->type != TL_WRITE_ALLOW_WRITE ||
512
620
              !lock->write_wait.data ||
514
622
            break;
515
623
          data=lock->write_wait.data;           /* Free this too */
516
624
        }
517
 
        if (data->type >= TL_WRITE)
 
625
        if (data->type >= TL_WRITE_LOW_PRIORITY)
518
626
          goto end;
519
627
        /* Release possible read locks together with the write lock */
520
628
      }
525
633
                             data->type == TL_WRITE_ALLOW_WRITE));
526
634
    }
527
635
    else if (data &&
528
 
             (lock_type=data->type) <= TL_WRITE_CONCURRENT_INSERT &&
 
636
             (lock_type=data->type) <= TL_WRITE_DELAYED &&
529
637
             ((lock_type != TL_WRITE_CONCURRENT_INSERT &&
530
638
               lock_type != TL_WRITE_ALLOW_WRITE) ||
531
639
              !lock->read_no_write_count))
532
640
    {
 
641
      /*
 
642
        For DELAYED, ALLOW_READ, WRITE_ALLOW_WRITE or CONCURRENT_INSERT locks
 
643
        start WRITE locks together with the READ locks
 
644
      */
 
645
      if (lock_type == TL_WRITE_CONCURRENT_INSERT &&
 
646
          (*lock->check_status)(data->status_param))
 
647
      {
 
648
        data->type=TL_WRITE;                    /* Upgrade lock */
 
649
        if (lock->read_wait.data)
 
650
          free_all_read_locks(lock,0);
 
651
        goto end;
 
652
      }
533
653
      do {
534
 
        boost::condition_variable_any *cond= data->cond;
 
654
        pthread_cond_t *cond=data->cond;
535
655
        if (((*data->prev)=data->next))         /* remove from wait-list */
536
656
          data->next->prev= data->prev;
537
657
        else
540
660
        data->prev=lock->write.last;
541
661
        lock->write.last= &data->next;
542
662
        data->next=0;                           /* Only one write lock */
543
 
        data->cond= NULL;                               /* Mark thread free */
544
 
        cond->notify_one(); /* Start waiting thread */
 
663
        data->cond=0;                           /* Mark thread free */
 
664
        pthread_cond_signal(cond);      /* Start waiting thread */
545
665
      } while (lock_type == TL_WRITE_ALLOW_WRITE &&
546
666
               (data=lock->write_wait.data) &&
547
667
               data->type == TL_WRITE_ALLOW_WRITE);
551
671
                             lock_type == TL_WRITE_ALLOW_WRITE));
552
672
    }
553
673
    else if (!data && lock->read_wait.data)
554
 
    {
555
674
      free_all_read_locks(lock,0);
556
 
    }
557
675
  }
558
676
end:
559
677
  return;
591
709
 
592
710
 
593
711
enum enum_thr_lock_result
594
 
thr_multi_lock(Session &session, THR_LOCK_DATA **data, uint32_t count, THR_LOCK_OWNER *owner)
 
712
thr_multi_lock(THR_LOCK_DATA **data, uint32_t count, THR_LOCK_OWNER *owner)
595
713
{
596
714
  THR_LOCK_DATA **pos,**end;
597
715
  if (count > 1)
599
717
  /* lock everything */
600
718
  for (pos=data,end=data+count; pos < end ; pos++)
601
719
  {
602
 
    enum enum_thr_lock_result result= thr_lock(session, *pos, owner, (*pos)->type);
 
720
    enum enum_thr_lock_result result= thr_lock(*pos, owner, (*pos)->type);
603
721
    if (result != THR_LOCK_SUCCESS)
604
722
    {                                           /* Aborted */
605
723
      thr_multi_unlock(data,(uint32_t) (pos-data));
619
737
    do
620
738
    {
621
739
      pos--;
622
 
      last_lock=(*pos);
 
740
      if (last_lock->lock == (*pos)->lock &&
 
741
          last_lock->lock->copy_status)
 
742
      {
 
743
        if (last_lock->type <= TL_READ_NO_INSERT)
 
744
        {
 
745
          THR_LOCK_DATA **read_lock;
 
746
          /*
 
747
            If we are locking the same table with read locks we must ensure
 
748
            that all tables share the status of the last write lock or
 
749
            the same read lock.
 
750
          */
 
751
          for (;
 
752
               (*pos)->type <= TL_READ_NO_INSERT &&
 
753
                 pos != data &&
 
754
                 pos[-1]->lock == (*pos)->lock ;
 
755
               pos--) ;
 
756
 
 
757
          read_lock = pos+1;
 
758
          do
 
759
          {
 
760
            (last_lock->lock->copy_status)((*read_lock)->status_param,
 
761
                                           (*pos)->status_param);
 
762
          } while (*(read_lock++) != last_lock);
 
763
          last_lock= (*pos);                    /* Point at last write lock */
 
764
        }
 
765
        else
 
766
          (*last_lock->lock->copy_status)((*pos)->status_param,
 
767
                                          last_lock->status_param);
 
768
      }
 
769
      else
 
770
        last_lock=(*pos);
623
771
    } while (pos != data);
624
772
  }
625
773
#endif
640
788
  return;
641
789
}
642
790
 
643
 
void DrizzleLock::unlock(uint32_t count)
644
 
{
645
 
  THR_LOCK_DATA **pos,**end;
646
 
 
647
 
  for (pos= getLocks(),end= getLocks()+count; pos < end ; pos++)
648
 
  {
649
 
    if ((*pos)->type != TL_UNLOCK)
650
 
      thr_unlock(*pos);
651
 
  }
652
 
}
653
 
 
654
791
/*
655
792
  Abort all threads waiting for a lock. The lock will be upgraded to
656
793
  TL_WRITE_ONLY to abort any new accesses to the lock
657
794
*/
658
795
 
659
 
void THR_LOCK::abort_locks()
 
796
void thr_abort_locks(THR_LOCK *lock, bool upgrade_lock)
660
797
{
661
 
  boost_unique_lock_t scopedLock(mutex);
 
798
  THR_LOCK_DATA *data;
 
799
  pthread_mutex_lock(&lock->mutex);
662
800
 
663
 
  for (THR_LOCK_DATA *local_data= read_wait.data; local_data ; local_data= local_data->next)
 
801
  for (data=lock->read_wait.data; data ; data=data->next)
664
802
  {
665
 
    local_data->type= TL_UNLOCK;                        /* Mark killed */
 
803
    data->type=TL_UNLOCK;                       /* Mark killed */
666
804
    /* It's safe to signal the cond first: we're still holding the mutex. */
667
 
    local_data->cond->notify_one();
668
 
    local_data->cond= NULL;                             /* Removed from list */
 
805
    pthread_cond_signal(data->cond);
 
806
    data->cond=0;                               /* Removed from list */
669
807
  }
670
 
  for (THR_LOCK_DATA *local_data= write_wait.data; local_data ; local_data= local_data->next)
 
808
  for (data=lock->write_wait.data; data ; data=data->next)
671
809
  {
672
 
    local_data->type= TL_UNLOCK;
673
 
    local_data->cond->notify_one();
674
 
    local_data->cond= NULL;
 
810
    data->type=TL_UNLOCK;
 
811
    pthread_cond_signal(data->cond);
 
812
    data->cond=0;
675
813
  }
676
 
  read_wait.last= &read_wait.data;
677
 
  write_wait.last= &write_wait.data;
678
 
  read_wait.data= write_wait.data=0;
679
 
  if (write.data)
680
 
    write.data->type=TL_WRITE_ONLY;
 
814
  lock->read_wait.last= &lock->read_wait.data;
 
815
  lock->write_wait.last= &lock->write_wait.data;
 
816
  lock->read_wait.data=lock->write_wait.data=0;
 
817
  if (upgrade_lock && lock->write.data)
 
818
    lock->write.data->type=TL_WRITE_ONLY;
 
819
  pthread_mutex_unlock(&lock->mutex);
 
820
  return;
681
821
}
682
822
 
683
823
 
687
827
  This is used to abort all locks for a specific thread
688
828
*/
689
829
 
690
 
bool THR_LOCK::abort_locks_for_thread(uint64_t thread_id_arg)
 
830
bool thr_abort_locks_for_thread(THR_LOCK *lock, my_thread_id thread_id)
691
831
{
 
832
  THR_LOCK_DATA *data;
692
833
  bool found= false;
693
834
 
694
 
  boost_unique_lock_t scopedLock(mutex);
695
 
  for (THR_LOCK_DATA *local_data= read_wait.data; local_data ; local_data= local_data->next)
 
835
  pthread_mutex_lock(&lock->mutex);
 
836
  for (data= lock->read_wait.data; data ; data= data->next)
696
837
  {
697
 
    if (local_data->owner->info->thread_id == thread_id_arg)
 
838
    if (data->owner->info->thread_id == thread_id)    /* purecov: tested */
698
839
    {
699
 
      local_data->type= TL_UNLOCK;                      /* Mark killed */
 
840
      data->type= TL_UNLOCK;                    /* Mark killed */
700
841
      /* It's safe to signal the cond first: we're still holding the mutex. */
701
842
      found= true;
702
 
      local_data->cond->notify_one();
703
 
      local_data->cond= 0;                              /* Removed from list */
 
843
      pthread_cond_signal(data->cond);
 
844
      data->cond= 0;                            /* Removed from list */
704
845
 
705
 
      if (((*local_data->prev)= local_data->next))
706
 
        local_data->next->prev= local_data->prev;
 
846
      if (((*data->prev)= data->next))
 
847
        data->next->prev= data->prev;
707
848
      else
708
 
        read_wait.last= local_data->prev;
 
849
        lock->read_wait.last= data->prev;
709
850
    }
710
851
  }
711
 
  for (THR_LOCK_DATA *local_data= write_wait.data; local_data ; local_data= local_data->next)
 
852
  for (data= lock->write_wait.data; data ; data= data->next)
712
853
  {
713
 
    if (local_data->owner->info->thread_id == thread_id_arg)
 
854
    if (data->owner->info->thread_id == thread_id) /* purecov: tested */
714
855
    {
715
 
      local_data->type= TL_UNLOCK;
 
856
      data->type= TL_UNLOCK;
716
857
      found= true;
717
 
      local_data->cond->notify_one();
718
 
      local_data->cond= NULL;
 
858
      pthread_cond_signal(data->cond);
 
859
      data->cond= 0;
719
860
 
720
 
      if (((*local_data->prev)= local_data->next))
721
 
        local_data->next->prev= local_data->prev;
 
861
      if (((*data->prev)= data->next))
 
862
        data->next->prev= data->prev;
722
863
      else
723
 
        write_wait.last= local_data->prev;
724
 
    }
725
 
  }
726
 
  wake_up_waiters(this);
727
 
 
728
 
  return found;
729
 
}
730
 
 
731
 
} /* namespace drizzled */
 
864
        lock->write_wait.last= data->prev;
 
865
    }
 
866
  }
 
867
  wake_up_waiters(lock);
 
868
  pthread_mutex_unlock(&lock->mutex);
 
869
  return(found);
 
870
}
 
871
 
 
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
}