00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00079 #include <config.h>
00080
00081 #include <fcntl.h>
00082
00083 #include <drizzled/error.h>
00084 #include <drizzled/thr_lock.h>
00085 #include <drizzled/session.h>
00086 #include <drizzled/session/times.h>
00087 #include <drizzled/sql_base.h>
00088 #include <drizzled/lock.h>
00089 #include <drizzled/pthread_globals.h>
00090 #include <drizzled/internal/my_sys.h>
00091 #include <drizzled/pthread_globals.h>
00092 #include <drizzled/plugin/storage_engine.h>
00093 #include <drizzled/util/test.h>
00094 #include <drizzled/open_tables_state.h>
00095 #include <drizzled/table/cache.h>
00096
00097 #include <set>
00098 #include <vector>
00099 #include <algorithm>
00100 #include <functional>
00101
00102 #include <boost/thread/shared_mutex.hpp>
00103 #include <boost/thread/condition_variable.hpp>
00104
00105 using namespace std;
00106
00107 namespace drizzled
00108 {
00109
00110 static boost::mutex LOCK_global_read_lock;
00111 static boost::condition_variable_any COND_global_read_lock;
00112
00118 static void print_lock_error(int error, const char *);
00119
00120
00121
00122
00123
00124
00125
00126
00127
00128
00129
00130
00131
00132
00133
00134
00135
00136
00137
00138
00139
00140
00141
00142 static drizzled::error_t thr_lock_errno_to_mysql[]=
00143 { EE_OK, EE_ERROR_FIRST, ER_LOCK_WAIT_TIMEOUT, ER_LOCK_DEADLOCK };
00144
00145
00166 static void reset_lock_data_and_free(DrizzleLock*& lock)
00167 {
00168 lock->reset();
00169 delete lock;
00170 lock= NULL;
00171 }
00172
00173 DrizzleLock *Session::lockTables(Table **tables, uint32_t count, uint32_t flags)
00174 {
00175 DrizzleLock *sql_lock;
00176 Table *write_lock_used;
00177 vector<plugin::StorageEngine *> involved_engines;
00178
00179 do
00180 {
00181 if (! (sql_lock= get_lock_data(tables, count, true, &write_lock_used)))
00182 break;
00183
00184 if (global_read_lock && write_lock_used and (not (flags & DRIZZLE_LOCK_IGNORE_GLOBAL_READ_LOCK)))
00185 {
00186
00187
00188
00189
00190 if (wait_if_global_read_lock(1, 1))
00191 {
00192
00193 reset_lock_data_and_free(sql_lock);
00194 break;
00195 }
00196
00197 if (open_tables.version != g_refresh_version)
00198 {
00199
00200 reset_lock_data_and_free(sql_lock);
00201 break;
00202 }
00203 }
00204
00205 set_proc_info("Notify start statement");
00206
00207
00208
00209
00210 if (sql_lock->sizeTable())
00211 {
00212 size_t num_tables= sql_lock->sizeTable();
00213 plugin::StorageEngine *engine;
00214 std::set<size_t> involved_slots;
00215
00216 for (size_t x= 1; x <= num_tables; x++, tables++)
00217 {
00218 engine= (*tables)->cursor->getEngine();
00219
00220 if (involved_slots.count(engine->getId()) > 0)
00221 continue;
00222
00223 involved_engines.push_back(engine);
00224 involved_slots.insert(engine->getId());
00225 }
00226
00227 for_each(involved_engines.begin(),
00228 involved_engines.end(),
00229 bind2nd(mem_fun(&plugin::StorageEngine::startStatement), this));
00230 }
00231
00232 set_proc_info("External lock");
00233
00234
00235
00236
00237
00238
00239 if (sql_lock->sizeTable() && lock_external(sql_lock->getTable(), sql_lock->sizeTable()))
00240 {
00241
00242 reset_lock_data_and_free(sql_lock);
00243 break;
00244 }
00245 set_proc_info("Table lock");
00246
00247 memcpy(sql_lock->getLocks() + sql_lock->sizeLock(),
00248 sql_lock->getLocks(),
00249 sql_lock->sizeLock() * sizeof(*sql_lock->getLocks()));
00250
00251
00252 drizzled::error_t rc;
00253 rc= thr_lock_errno_to_mysql[(int) thr_multi_lock(*this,
00254 sql_lock->getLocks() +
00255 sql_lock->sizeLock(),
00256 sql_lock->sizeLock(),
00257 this->lock_id)];
00258 if (rc)
00259 {
00260 if (sql_lock->sizeTable())
00261 unlock_external(sql_lock->getTable(), sql_lock->sizeTable());
00262 reset_lock_data_and_free(sql_lock);
00263 my_error(rc, MYF(0));
00264 }
00265 } while(0);
00266
00267 set_proc_info(0);
00268 if (getKilled())
00269 {
00270 send_kill_message();
00271 if (sql_lock)
00272 {
00273 unlockTables(sql_lock);
00274 sql_lock= NULL;
00275 }
00276 }
00277
00278 times.set_time_after_lock();
00279
00280 return sql_lock;
00281 }
00282
00283
00284 int Session::lock_external(Table **tables, uint32_t count)
00285 {
00286 int lock_type,error;
00287 for (uint32_t i= 1 ; i <= count ; i++, tables++)
00288 {
00289 assert((*tables)->reginfo.lock_type >= TL_READ);
00290 lock_type=F_WRLCK;
00291 if ((*tables)->db_stat & HA_READ_ONLY ||
00292 ((*tables)->reginfo.lock_type >= TL_READ &&
00293 (*tables)->reginfo.lock_type <= TL_READ_NO_INSERT))
00294 lock_type=F_RDLCK;
00295
00296 if ((error=(*tables)->cursor->ha_external_lock(this,lock_type)))
00297 {
00298 print_lock_error(error, (*tables)->cursor->getEngine()->getName().c_str());
00299 while (--i)
00300 {
00301 tables--;
00302 (*tables)->cursor->ha_external_lock(this, F_UNLCK);
00303 (*tables)->current_lock=F_UNLCK;
00304 }
00305 return error;
00306 }
00307 else
00308 {
00309 (*tables)->db_stat &= ~ HA_BLOCK_LOCK;
00310 (*tables)->current_lock= lock_type;
00311 }
00312 }
00313 return 0;
00314 }
00315
00316
00317 void Session::unlockTables(DrizzleLock *sql_lock)
00318 {
00319 if (sql_lock->sizeLock())
00320 sql_lock->unlock(sql_lock->sizeLock());
00321 if (sql_lock->sizeTable())
00322 unlock_external(sql_lock->getTable(), sql_lock->sizeTable());
00323 delete sql_lock;
00324 }
00325
00332 void Session::unlockSomeTables(Table **table, uint32_t count)
00333 {
00334 DrizzleLock *sql_lock;
00335 Table *write_lock_used;
00336 if ((sql_lock= get_lock_data(table, count, false,
00337 &write_lock_used)))
00338 unlockTables(sql_lock);
00339 }
00340
00341
00346 void Session::unlockReadTables(DrizzleLock *sql_lock)
00347 {
00348 uint32_t i,found;
00349
00350
00351 THR_LOCK_DATA **lock_local= sql_lock->getLocks();
00352 for (i=found=0 ; i < sql_lock->sizeLock(); i++)
00353 {
00354 if (sql_lock->getLocks()[i]->type >= TL_WRITE_ALLOW_READ)
00355 {
00356 std::swap(*lock_local, sql_lock->getLocks()[i]);
00357 lock_local++;
00358 found++;
00359 }
00360 }
00361
00362 if (i != found)
00363 {
00364 thr_multi_unlock(lock_local, i - found);
00365 sql_lock->setLock(found);
00366 }
00367
00368
00369
00370 Table **table= sql_lock->getTable();
00371 for (i=found=0 ; i < sql_lock->sizeTable() ; i++)
00372 {
00373 assert(sql_lock->getTable()[i]->lock_position == i);
00374 if ((uint32_t) sql_lock->getTable()[i]->reginfo.lock_type >= TL_WRITE_ALLOW_READ)
00375 {
00376 std::swap(*table, sql_lock->getTable()[i]);
00377 table++;
00378 found++;
00379 }
00380 }
00381
00382 if (i != found)
00383 {
00384 unlock_external(table, i - found);
00385 sql_lock->resizeTable(found);
00386 }
00387
00388 table= sql_lock->getTable();
00389 found= 0;
00390 for (i= 0; i < sql_lock->sizeTable(); i++)
00391 {
00392 Table *tbl= *table;
00393 tbl->lock_position= table - sql_lock->getTable();
00394 tbl->lock_data_start= found;
00395 found+= tbl->lock_count;
00396 table++;
00397 }
00398 }
00399
00400
00421 void Session::removeLock(Table *table)
00422 {
00423 unlockSomeTables(&table, 1);
00424 }
00425
00426
00429 void Session::abortLock(Table *table)
00430 {
00431 DrizzleLock *locked;
00432 Table *write_lock_used;
00433
00434 if ((locked= get_lock_data(&table, 1, false,
00435 &write_lock_used)))
00436 {
00437 for (uint32_t x= 0; x < locked->sizeLock(); x++)
00438 locked->getLocks()[x]->lock->abort_locks();
00439 delete locked;
00440 }
00441 }
00442
00443
00456 bool Session::abortLockForThread(Table *table)
00457 {
00458 bool result= false;
00459 Table* write_lock_used;
00460 if (DrizzleLock* locked= get_lock_data(&table, 1, false, &write_lock_used))
00461 {
00462 for (uint32_t i= 0; i < locked->sizeLock(); i++)
00463 {
00464 if (locked->getLocks()[i]->lock->abort_locks_for_thread(table->in_use->thread_id))
00465 result= true;
00466 }
00467 delete locked;
00468 }
00469 return result;
00470 }
00471
00474 int Session::unlock_external(Table **table, uint32_t count)
00475 {
00476 int error;
00477
00478 int error_code=0;
00479 do
00480 {
00481 if ((*table)->current_lock != F_UNLCK)
00482 {
00483 (*table)->current_lock = F_UNLCK;
00484 if ((error=(*table)->cursor->ha_external_lock(this, F_UNLCK)))
00485 {
00486 error_code=error;
00487 print_lock_error(error_code, (*table)->cursor->getEngine()->getName().c_str());
00488 }
00489 }
00490 table++;
00491 } while (--count);
00492 return error_code;
00493 }
00494
00495
00507 DrizzleLock *Session::get_lock_data(Table **table_ptr, uint32_t count,
00508 bool should_lock, Table **write_lock_used)
00509 {
00510 uint32_t lock_count;
00511 THR_LOCK_DATA **locks, **locks_buf, **locks_start;
00512 Table **to, **table_buf;
00513
00514 *write_lock_used=0;
00515 for (uint32_t i= lock_count= 0 ; i < count ; i++)
00516 {
00517 Table *t= table_ptr[i];
00518
00519 if (! (t->getEngine()->check_flag(HTON_BIT_SKIP_STORE_LOCK)))
00520 {
00521 lock_count++;
00522 }
00523 }
00524
00525
00526
00527
00528
00529
00530
00531 DrizzleLock *sql_lock= new DrizzleLock(lock_count);
00532
00533 if (not sql_lock)
00534 return NULL;
00535
00536 locks= locks_buf= sql_lock->getLocks();
00537 to= table_buf= sql_lock->getTable();
00538
00539 for (uint32_t i= 0; i < count ; i++)
00540 {
00541 Table *table;
00542 thr_lock_type lock_type;
00543
00544 if (table_ptr[i]->getEngine()->check_flag(HTON_BIT_SKIP_STORE_LOCK))
00545 continue;
00546
00547 table= table_ptr[i];
00548 lock_type= table->reginfo.lock_type;
00549 assert (lock_type != TL_WRITE_DEFAULT);
00550 if (lock_type >= TL_WRITE_ALLOW_WRITE)
00551 {
00552 *write_lock_used=table;
00553 if (table->db_stat & HA_READ_ONLY)
00554 {
00555 my_error(ER_OPEN_AS_READONLY, MYF(0), table->getAlias());
00556
00557 sql_lock->setLock(locks - sql_lock->getLocks());
00558 reset_lock_data_and_free(sql_lock);
00559 return NULL;
00560 }
00561 }
00562 locks_start= locks;
00563 locks= table->cursor->store_lock(this, locks, should_lock ? lock_type : TL_IGNORE);
00564 if (should_lock)
00565 {
00566 table->lock_position= (uint32_t) (to - table_buf);
00567 table->lock_data_start= (uint32_t) (locks_start - locks_buf);
00568 table->lock_count= (uint32_t) (locks - locks_start);
00569 assert(table->lock_count == 1);
00570 }
00571 *to++= table;
00572 }
00573
00574
00575
00576
00577
00578
00579
00580
00581
00582
00583
00584
00585
00586
00587 sql_lock->setLock(locks - locks_buf);
00588
00589 return sql_lock;
00590 }
00591
00592
00619 int Session::lock_table_name(TableList *table_list)
00620 {
00621 identifier::Table identifier(table_list->getSchemaName(), table_list->getTableName());
00622 {
00623
00624 table::CacheRange ppp= table::getCache().equal_range(identifier.getKey());
00625 for (table::CacheMap::const_iterator iter= ppp.first; iter != ppp.second; ++iter)
00626 {
00627 Table *table= iter->second;
00628 if (table->reginfo.lock_type < TL_WRITE)
00629 continue;
00630 if (table->in_use == this)
00631 {
00632 table->getMutableShare()->resetVersion();
00633 table->locked_by_name= true;
00634 return 0;
00635 }
00636 }
00637 }
00638
00639 table::Placeholder *table= &table_cache_insert_placeholder(identifier);
00640 table_list->table= reinterpret_cast<Table*>(table);
00641
00642
00643 return (test(table::Cache::removeTable(*this, identifier, RTFC_NO_FLAG)));
00644 }
00645
00646
00647 void TableList::unlock_table_name()
00648 {
00649 if (table)
00650 {
00651 table::remove_table(static_cast<table::Concurrent *>(table));
00652 locking::broadcast_refresh();
00653 }
00654 }
00655
00656
00657 static bool locked_named_table(TableList *table_list)
00658 {
00659 for (; table_list; table_list=table_list->next_local)
00660 {
00661 Table *table= table_list->table;
00662 if (table)
00663 {
00664 Table *save_next= table->getNext();
00665 table->setNext(NULL);
00666 bool result= table::Cache::areTablesUsed(table_list->table, 0);
00667 table->setNext(save_next);
00668 if (result)
00669 return 1;
00670 }
00671 }
00672 return 0;
00673 }
00674
00675
00676 bool Session::wait_for_locked_table_names(TableList *table_list)
00677 {
00678 bool result= false;
00679
00680 #if 0
00681 assert(ownership of table::Cache::mutex());
00682 #endif
00683
00684 while (locked_named_table(table_list))
00685 {
00686 if (getKilled())
00687 {
00688 result= true;
00689 break;
00690 }
00691 wait_for_condition(table::Cache::mutex(), COND_refresh);
00692 table::Cache::mutex().lock();
00693 }
00694 return result;
00695 }
00696
00697
00712 bool Session::lock_table_names(TableList *table_list)
00713 {
00714 bool got_all_locks= true;
00715 for (TableList* lock_table= table_list; lock_table; lock_table= lock_table->next_local)
00716 {
00717 int got_lock= lock_table_name(lock_table);
00718 if (got_lock < 0)
00719 {
00720 table_list->unlock_table_names(table_list);
00721 return true;
00722 }
00723 if (got_lock)
00724 got_all_locks= false;
00725 }
00726
00727
00728 if (not got_all_locks && wait_for_locked_table_names(table_list))
00729 {
00730 table_list->unlock_table_names(table_list);
00731 return true;
00732 }
00733 return false;
00734 }
00735
00736
00755 bool Session::lock_table_names_exclusively(TableList *table_list)
00756 {
00757 if (lock_table_names(table_list))
00758 return true;
00759
00760
00761
00762
00763 for (TableList *table= table_list; table; table= table->next_global)
00764 {
00765 if (table->table)
00766 table->table->open_placeholder= 1;
00767 }
00768 return false;
00769 }
00770
00771
00794 void TableList::unlock_table_names(TableList *last_table)
00795 {
00796 for (TableList *table_iter= this; table_iter != last_table; table_iter= table_iter->next_local)
00797 {
00798 table_iter->unlock_table_name();
00799 }
00800 locking::broadcast_refresh();
00801 }
00802
00803
00804 static void print_lock_error(int error, const char *table)
00805 {
00806 drizzled::error_t textno;
00807 switch (error)
00808 {
00809 case HA_ERR_LOCK_WAIT_TIMEOUT:
00810 textno=ER_LOCK_WAIT_TIMEOUT;
00811 break;
00812 case HA_ERR_READ_ONLY_TRANSACTION:
00813 textno=ER_READ_ONLY_TRANSACTION;
00814 break;
00815 case HA_ERR_LOCK_DEADLOCK:
00816 textno=ER_LOCK_DEADLOCK;
00817 break;
00818 case HA_ERR_WRONG_COMMAND:
00819 textno=ER_ILLEGAL_HA;
00820 break;
00821 default:
00822 textno=ER_CANT_LOCK;
00823 break;
00824 }
00825
00826 if ( textno == ER_ILLEGAL_HA )
00827 my_error(textno, MYF(ME_BELL+ME_OLDWIN+ME_WAITTANG), table);
00828 else
00829 my_error(textno, MYF(ME_BELL+ME_OLDWIN+ME_WAITTANG), error);
00830 }
00831
00832
00833
00834
00835
00836
00837
00838
00839
00840
00841
00842
00843
00844
00845
00846
00847
00848
00849
00850
00851
00852
00853
00854
00855
00856
00857
00858
00859
00860
00861
00862
00863
00864
00865
00866
00867
00868
00869
00870
00871
00872
00873
00874
00875
00876
00877
00878
00879
00880
00881
00882
00883
00884
00885
00886
00887
00888
00889
00890
00891
00892
00893
00894
00895
00896
00897
00898
00899
00900
00901
00902
00903
00904
00905
00906
00907
00908 volatile uint32_t global_read_lock=0;
00909 volatile uint32_t global_read_lock_blocks_commit=0;
00910 static volatile uint32_t protect_against_global_read_lock=0;
00911 static volatile uint32_t waiting_for_read_lock=0;
00912
00913 bool Session::lockGlobalReadLock()
00914 {
00915 if (isGlobalReadLock() == Session::NONE)
00916 {
00917 const char *old_message;
00918 LOCK_global_read_lock.lock();
00919 old_message= enter_cond(COND_global_read_lock, LOCK_global_read_lock,
00920 "Waiting to get readlock");
00921
00922 waiting_for_read_lock++;
00923 boost::mutex::scoped_lock scopedLock(LOCK_global_read_lock, boost::adopt_lock_t());
00924 while (protect_against_global_read_lock && not getKilled())
00925 COND_global_read_lock.wait(scopedLock);
00926 waiting_for_read_lock--;
00927 scopedLock.release();
00928 if (getKilled())
00929 {
00930 exit_cond(old_message);
00931 return true;
00932 }
00933 setGlobalReadLock(Session::GOT_GLOBAL_READ_LOCK);
00934 global_read_lock++;
00935 exit_cond(old_message);
00936 }
00937
00938
00939
00940
00941
00942
00943
00944
00945
00946 return false;
00947 }
00948
00949
00950 void Session::unlockGlobalReadLock(void)
00951 {
00952 uint32_t tmp;
00953
00954 if (not isGlobalReadLock())
00955 return;
00956
00957 {
00958 boost::mutex::scoped_lock scopedLock(LOCK_global_read_lock);
00959 tmp= --global_read_lock;
00960 if (isGlobalReadLock() == Session::MADE_GLOBAL_READ_LOCK_BLOCK_COMMIT)
00961 --global_read_lock_blocks_commit;
00962 }
00963
00964 if (not tmp)
00965 {
00966 COND_global_read_lock.notify_all();
00967 }
00968 setGlobalReadLock(Session::NONE);
00969 }
00970
00971 static inline bool must_wait(bool is_not_commit)
00972 {
00973 return (global_read_lock &&
00974 (is_not_commit ||
00975 global_read_lock_blocks_commit));
00976 }
00977
00978 bool Session::wait_if_global_read_lock(bool abort_on_refresh, bool is_not_commit)
00979 {
00980 const char *old_message= NULL;
00981 bool result= 0, need_exit_cond;
00982
00983
00984
00985
00986
00987
00988 safe_mutex_assert_not_owner(table::Cache::mutex().native_handle());
00989
00990 LOCK_global_read_lock.lock();
00991 if ((need_exit_cond= must_wait(is_not_commit)))
00992 {
00993 if (isGlobalReadLock())
00994 {
00995 if (is_not_commit)
00996 my_message(ER_CANT_UPDATE_WITH_READLOCK,
00997 ER(ER_CANT_UPDATE_WITH_READLOCK), MYF(0));
00998 LOCK_global_read_lock.unlock();
00999
01000
01001
01002
01003
01004 return is_not_commit;
01005 }
01006 old_message= enter_cond(COND_global_read_lock, LOCK_global_read_lock,
01007 "Waiting for release of readlock");
01008
01009 while (must_wait(is_not_commit) && not getKilled() &&
01010 (!abort_on_refresh || open_tables.version == g_refresh_version))
01011 {
01012 boost::mutex::scoped_lock scoped(LOCK_global_read_lock, boost::adopt_lock_t());
01013 COND_global_read_lock.wait(scoped);
01014 scoped.release();
01015 }
01016 if (getKilled())
01017 result=1;
01018 }
01019
01020 if (not abort_on_refresh && not result)
01021 protect_against_global_read_lock++;
01022
01023
01024
01025
01026
01027 if (unlikely(need_exit_cond))
01028 {
01029 exit_cond(old_message);
01030 }
01031 else
01032 {
01033 LOCK_global_read_lock.unlock();
01034 }
01035
01036 return result;
01037 }
01038
01039
01040 void Session::startWaitingGlobalReadLock()
01041 {
01042 if (unlikely(isGlobalReadLock()))
01043 return;
01044
01045 LOCK_global_read_lock.lock();
01046 bool tmp= (!--protect_against_global_read_lock && (waiting_for_read_lock || global_read_lock_blocks_commit));
01047 LOCK_global_read_lock.unlock();
01048
01049 if (tmp)
01050 COND_global_read_lock.notify_all();
01051 }
01052
01053
01054 bool Session::makeGlobalReadLockBlockCommit()
01055 {
01056 bool error;
01057 const char *old_message;
01058
01059
01060
01061
01062 if (isGlobalReadLock() != Session::GOT_GLOBAL_READ_LOCK)
01063 return false;
01064 LOCK_global_read_lock.lock();
01065
01066 global_read_lock_blocks_commit++;
01067 old_message= enter_cond(COND_global_read_lock, LOCK_global_read_lock,
01068 "Waiting for all running commits to finish");
01069 while (protect_against_global_read_lock && not getKilled())
01070 {
01071 boost::mutex::scoped_lock scopedLock(LOCK_global_read_lock, boost::adopt_lock_t());
01072 COND_global_read_lock.wait(scopedLock);
01073 scopedLock.release();
01074 }
01075 if ((error= test(getKilled())))
01076 {
01077 global_read_lock_blocks_commit--;
01078 }
01079 else
01080 {
01081 setGlobalReadLock(Session::MADE_GLOBAL_READ_LOCK_BLOCK_COMMIT);
01082 }
01083
01084 exit_cond(old_message);
01085
01086 return error;
01087 }
01088
01089
01109 void locking::broadcast_refresh()
01110 {
01111 COND_refresh.notify_all();
01112 COND_global_read_lock.notify_all();
01113 }
01114
01115 }