Bug #13955256: KEYCACHE CRASHES, CORRUPTIONS/HANGS WITH,

               FULLTEXT INDEX AND CONCURRENT DML.

Problem Statement:
------------------
1) Create a table with FT index.
2) Enable concurrent inserts.
3) In multiple threads do below operations repeatedly
   a) truncate table
   b) insert into table ....
   c) select ... match .. against .. non-boolean/boolean mode

After some time we could observe two different assert core dumps

Analysis:
--------
1)assert core dump at key_read_cache():
Two select threads operating in-parallel on same key 
root block.
1st select thread block->status is set to BLOCK_ERROR 
because the my_pread() in read_block() is returning '0'. 
Truncate table made the index file size as 1024 and pread 
was asked to get the block of count bytes(1024 bytes) 
from offset of 1024 which it cannot read since its 
"end of file" and retuning '0' setting 
"my_errno= HA_ERR_FILE_TOO_SHORT" and the key_file_length, 
key_root[0] is same i.e. 1024. Since block status has BLOCK_ERROR 
the 1st select thread enter into the free_block() and will 
be under wait on conditional mutex by making status as 
BLOCK_REASSIGNED and goes for wait_on_readers(). Other select 
thread will also work on the same block and sees the status as 
BLOCK_ERROR and enters into free_block(), checks for BLOCK_REASSIGNED 
and asserting the server.

2)assert core dump at key_write_cache():
One select thread and One insert thread.
Select thread gets the unlocks the 'keycache->cache_lock', 
which allows other threads to continue and gets the pread() 
return value as'0'(please see the explanation above) and 
tries to get the lock on 'keycache->cache_lock' and waits 
there for the lock.
Insert thread requests for the block, block will be assigned 
from the hash list and makes the page_status as 
'PAGE_WAIT_TO_BE_READ' and goes for the read_block(), waits 
in the queue since there are some other threads performing 
reads on the same block.
Select thread which was waiting for the 'keycache->cache_lock' 
mutex in the read_block() will continue after getting the my_pread() 
value as '0' and sets the block status as BLOCK_ERROR and goes to 
the free_block() and go to the wait_for_readers().
Now the insert thread will awake and continues. and checks 
block->status as not BLOCK_READ and it asserts.  

Fix:
---
In the full text code, multiple readers of index file is not guarded. 
Hence added below below code in _ft2_search() and walk_and_match().

to lock the key_root I have used below code in _ft2_search()
 if (info->s->concurrent_insert)
    mysql_rwlock_rdlock(&share->key_root_lock[0]);

and to unlock 
 if (info->s->concurrent_insert)
   mysql_rwlock_unlock(&share->key_root_lock[0]);

storage/myisam/ft_boolean_search.c:
  Since its a recursion function, to avoid confusion in taking and 
  releasing the locks, renamed _ft2_search() to _ft2_search_internal() 
  function. And _ft2_search() will take the lock, call 
  _ft2_search_internal() and release the lock in case of concurrent 
  inserts.
storage/myisam/ft_nlq_search.c:
  Added read locks code in walk_and_match()

2 files changed, 31 insertions, 3 deletions

diff --git a/storage/myisam/ft_boolean_search.c b/storage/myisam/ft_boolean_search.c
index 62d912a0fd0..16cf2669c6d 100644
--- a/storage/myisam/ft_boolean_search.c
+++ b/storage/myisam/ft_boolean_search.c
@@ -354,7 +354,7 @@ static int _ftb_no_dupes_cmp(void* not_used __attribute__((unused)),
 
   returns 1 if the search was finished (must-word wasn't found)
 */
-static int _ft2_search(FTB *ftb, FTB_WORD *ftbw, my_bool init_search)
+static int _ft2_search_no_lock(FTB *ftb, FTB_WORD *ftbw, my_bool init_search)
 {
   int r;
   int subkeys=1;
@@ -454,7 +454,7 @@ static int _ft2_search(FTB *ftb, FTB_WORD *ftbw, my_bool init_search)
     ftbw->key_root=info->s->state.key_root[ftb->keynr];
     ftbw->keyinfo=info->s->keyinfo+ftb->keynr;
     ftbw->off=0;
-    return _ft2_search(ftb, ftbw, 0);
+    return _ft2_search_no_lock(ftb, ftbw, 0);
   }
 
   /* matching key found */
@@ -482,6 +482,18 @@ static int _ft2_search(FTB *ftb, FTB_WORD *ftbw, my_bool init_search)
   return 0;
 }
 
+static int _ft2_search(FTB *ftb, FTB_WORD *ftbw, my_bool init_search)
+{
+  int r;
+  MYISAM_SHARE *share= ftb->info->s;
+  if (share->concurrent_insert)
+    rw_rdlock(&share->key_root_lock[ftb->keynr]);
+  r= _ft2_search_no_lock(ftb, ftbw, init_search);
+  if (share->concurrent_insert)
+    rw_unlock(&share->key_root_lock[ftb->keynr]);
+  return r;
+}
+
 static void _ftb_init_index_search(FT_INFO *ftb)
 {
   int i;
diff --git a/storage/myisam/ft_nlq_search.c b/storage/myisam/ft_nlq_search.c
index 333ec5d0b4f..c74451828a4 100644
--- a/storage/myisam/ft_nlq_search.c
+++ b/storage/myisam/ft_nlq_search.c
@@ -71,9 +71,10 @@ static int walk_and_match(FT_WORD *word, uint32 count, ALL_IN_ONE *aio)
   TREE_ELEMENT *selem;
   double       gweight=1;
   MI_INFO      *info=aio->info;
+  MYISAM_SHARE *share= info->s;
   uchar        *keybuff=aio->keybuff;
   MI_KEYDEF    *keyinfo=info->s->keyinfo+aio->keynr;
-  my_off_t     key_root=info->s->state.key_root[aio->keynr];
+  my_off_t     key_root;
   uint         extra= HA_FT_WLEN + info->s->rec_reflength;
 #if HA_FT_WTYPE == HA_KEYTYPE_FLOAT
   float tmp_weight;
@@ -89,6 +90,11 @@ static int walk_and_match(FT_WORD *word, uint32 count, ALL_IN_ONE *aio)
   keylen-=HA_FT_WLEN;
   doc_cnt=0;
 
+  if (share->concurrent_insert)
+    rw_rdlock(&share->key_root_lock[aio->keynr]);
+
+  key_root= share->state.key_root[aio->keynr];
+
   /* Skip rows inserted by current inserted */
   for (r=_mi_search(info, keyinfo, keybuff, keylen, SEARCH_FIND, key_root) ;
        !r &&
@@ -98,6 +104,9 @@ static int walk_and_match(FT_WORD *word, uint32 count, ALL_IN_ONE *aio)
                           info->lastkey_length, SEARCH_BIGGER, key_root))
     ;
 
+  if (share->concurrent_insert)
+    rw_unlock(&share->key_root_lock[aio->keynr]);
+
   info->update|= HA_STATE_AKTIV;              /* for _mi_test_if_changed() */
 
   /* The following should be safe, even if we compare doubles */
@@ -121,6 +130,8 @@ static int walk_and_match(FT_WORD *word, uint32 count, ALL_IN_ONE *aio)
       keyinfo=& info->s->ft2_keyinfo;
       key_root=info->lastpos;
       keylen=0;
+      if (share->concurrent_insert)
+        rw_rdlock(&share->key_root_lock[aio->keynr]);
       r=_mi_search_first(info, keyinfo, key_root);
       goto do_skip;
     }
@@ -155,6 +166,9 @@ static int walk_and_match(FT_WORD *word, uint32 count, ALL_IN_ONE *aio)
     if (gweight < 0 || doc_cnt > 2000000)
       gweight=0;
 
+    if (share->concurrent_insert)
+      rw_rdlock(&share->key_root_lock[aio->keynr]);
+
     if (_mi_test_if_changed(info) == 0)
 	r=_mi_search_next(info, keyinfo, info->lastkey, info->lastkey_length,
                           SEARCH_BIGGER, key_root);
@@ -167,6 +181,8 @@ do_skip:
       r= _mi_search_next(info, keyinfo, info->lastkey, info->lastkey_length,
                          SEARCH_BIGGER, key_root);
 
+    if (share->concurrent_insert)
+      rw_unlock(&share->key_root_lock[aio->keynr]);
   }
   word->weight=gweight;