[BUG]FlushMemTable races with each other and crashes under DEBUG mode

[wolfkdy]

this issue is duplicated with #5126
I open another ticket as the previous one is opened one year ago and seems untracked now as no one contributed a unittest.
this issue won't be a problem in release mode as asserts are eliminated.
The two asserts won't be held.

Status DBImpl::FlushMemTableToOutputFile(
    ColumnFamilyData* cfd, const MutableCFOptions& mutable_cf_options,
    bool* made_progress, JobContext* job_context,
    SuperVersionContext* superversion_context, LogBuffer* log_buffer) {
  mutex_.AssertHeld();
  assert(cfd->imm()->NumNotFlushed() != 0);
  assert(cfd->imm()->IsFlushPending());

Steps to reproduce the behavior

• flush thread 1 pick column family 0，1，2, 3 then flush 0 (release the mutex lock). This is constructed by Writing to all cfs and issuing a switch wal request.
• when thread 1 is flushing cf 0(with no mutex lock), user issue a Put and Flush to cf 1. let's assume flush thread 2 handles this request.
• flush thread 2 flushes all the memtables of cf 1 and returns OK.
• thread 1 is re-scheduled and tries to flush cf 1 and asserts.

these steps have been constributed into the unittest below, which can reproduce the two asserts.
The test works under 5.18.3, I guess it should also work under master branch.

TEST_F(DBFlushTest, ParallelFlush) {                                                                                
  Options options;                                                                                                  
  options.disable_auto_compactions = true;                                                                          
  options.max_background_flushes = 20;                                                                              
  options.env = env_;                                                                                               
  CreateAndReopenWithCF({"1", "2", "3"}, options);                                                                  
  FlushOptions flushOpt;                                                                                            
  flushOpt.wait = true;                                                                                             
  flushOpt.allow_write_stall=true;                                                                                  
                                                                                                                    
  bool triggered = false;                                                                                           
  SyncPoint::GetInstance()->SetCallBack(                                                                            
      "FlushJob::WriteLevel0Table", [&](void* /*arg*/) {                                                            
         if (triggered) return;                                                                                     
         triggered = true;                                                                                          
         ASSERT_OK(Put(static_cast<int>(1) /*cf*/, "key", "value"));                                                
         ASSERT_OK(dbfull()->Flush(flushOpt, handles_[1]));                                                         
      });                                                                                                           
  SyncPoint::GetInstance()->EnableProcessing();                                                                     
                                                                                                                    
                                                                                                                    
  for (size_t i = 0; i != handles_.size(); ++i) {                                                                   
    ASSERT_OK(Put(static_cast<int>(i) /*cf*/, "key", "value"));                                                     
  }                                                                                                                 
                                                                                                                    
  dbfull()->TEST_SwitchWAL();                                                                                       
  sleep(10);                                                                                                        
} 

[riversand963]

Thanks @wolfkdy for reporting. We can schedule multiple bg threads and each of them flushes one column family. Will submit a fix for this.

update
Changes required on flush scheduling seems more complex. Need to think more.

[riversand963]

@wolfkdy , given #7362 , I believe the bug is fixed. Let me know if not.