gbloader.cpp

来自「ncbi源码」· C++ 代码 · 共 1,566 行 · 第 1/4 页

                g.Lock(&*tse);
                {{ // make sure we have reverse reference to handle
                    STSEinfo::TSeqids &sid = tse->m_SeqIds;
                    if (sid.find(sih) == sid.end())
                        sid.insert(sih);
                }}
                
                if( !x_NeedMoreData(*tse) )
                    continue;
                // need update
                _ASSERT(tse->m_LoadState != STSEinfo::eLoadStateDone);
                tse->locked++;
                g.Local();

                CReader::TConn conn = m_Locks.m_Pool.Select(&*tse);
                try {
                    x_GetData(tse, conn);
                }
                catch ( CLoaderException& e ) {
                    g.Lock();
                    g.Lock(&*tse);
                    tse->locked--;
                    x_UpdateDropList(&*tse); // move up as just checked
                    switch ( e.GetErrCode() ) {
                    case CLoaderException::eNoConnection:
                        throw;
                    case CLoaderException::ePrivateData:
                        // no need to reconnect
                        // no need to wait more
                        break;
                    case CLoaderException::eNoData:
                        // no need to reconnect
                        throw;
                    default:
                        ERR_POST("GenBank connection failed: Reconnecting...");
                        m_Driver->Reconnect(conn);
                        throw;
                    }
                }
                catch ( ... ) {
                    g.Lock();
                    g.Lock(&*tse);
                    tse->locked--;
                    x_UpdateDropList(&*tse); // move up as just checked
                    ERR_POST("GenBank connection failed: Reconnecting...");
                    m_Driver->Reconnect(conn);
                    throw;
                }
                _ASSERT(!x_NeedMoreData(*tse));
                g.Lock();
                g.Lock(&*tse);
                tse->locked--;
                x_UpdateDropList(&*tse); // move up as just checked

                x_Check(&*tse);
                x_Check();
            }
            // everything is loaded, break the attempt loop
            break;
        }
        catch ( CLoaderException& e ) {
            ERR_POST(e.what());
            if ( e.GetErrCode() == e.eNoConnection ) {
                throw;
            }
        }
        catch ( exception& e ) {
            ERR_POST(e.what());
        }
        catch ( ... ) {
            ERR_POST(CThread::GetSelf()<<":: Data request failed....");
            throw;
        }
        // something is not loaded
        // in case of any error we'll force reloading seqrefs
        
        CGBLGuard g(m_Locks,CGBLGuard::eMain,"x_ResolveHandle");
        g.Lock();
        sr->m_Timer.Reset();
        m_Driver->PurgeSeqrefs(sr->m_Sr, *sih.GetSeqId());
    }
    if ( attempt >= attempt_count ) {
        ERR_POST("CGBLoader:GetData: data request failed: "
                 "exceeded maximum attempts count");
        NCBI_THROW(CLoaderException, eLoaderFailed,
                   "Multiple attempts to retrieve data failed");
    }
}


void CGBDataLoader::x_GetChunk(CRef<STSEinfo> tse,
                               CTSE_Chunk_Info& chunk_info)
{
    CGBLGuard g(m_Locks,"x_GetChunk");
    g.Lock(&*tse);
    tse->locked++;
    g.Local();

    bool done = false;
    int try_cnt = 3;
    while( !done && try_cnt-- > 0 ) {
        CReader::TConn conn = m_Locks.m_Pool.Select(&*tse);
        try {
            x_GetChunk(tse, conn, chunk_info);
            done = true;
            break;
        }
        catch ( CLoaderException& e ) {
            if ( e.GetErrCode() == CLoaderException::eNoConnection ) {
                g.Lock();
                g.Lock(&*tse);
                tse->locked--;
                x_UpdateDropList(&*tse); // move up as just checked
                throw;
            }
            LOG_POST(e.what());
            LOG_POST("GenBank connection failed: Reconnecting....");
            m_Driver->Reconnect(conn);
        }
        catch ( const exception &e ) {
            LOG_POST(e.what());
            LOG_POST("GenBank connection failed: Reconnecting....");
            m_Driver->Reconnect(conn);
        }
        catch ( ... ) {
            LOG_POST("GenBank connection failed: Reconnecting....");
            m_Driver->Reconnect(conn);
            g.Lock();
            g.Lock(&*tse);
            tse->locked--;
            x_UpdateDropList(&*tse); // move up as just checked
            throw;
        }
    }
    if ( !done ) {
        ERR_POST("CGBLoader:GetChunk: data request failed: "
                 "exceeded maximum attempts count");
        g.Lock();
        g.Lock(&*tse);
        tse->locked--;
        x_UpdateDropList(&*tse); // move up as just checked
        NCBI_THROW(CLoaderException, eLoaderFailed,
                   "Multiple attempts to retrieve data failed");
    }
    g.Lock();
    g.Lock(&*tse);
    x_UpdateDropList(&*tse);
    if( done ) {
        x_Check();
        _ASSERT(tse->tseinfop);
    }
    tse->locked--;
    x_Check(&*tse);
}


class CTimerGuard
{
    CTimer *t;
    bool    calibrating;
public:
    CTimerGuard(CTimer& x)
        : t(&x), calibrating(x.NeedCalibration())
        {
            if ( calibrating ) {
                t->Start();
            }
        }
    ~CTimerGuard(void)
        {
            if ( calibrating ) {
                t->Stop();
            }
        }
};


CRef<SSeqrefs> CGBDataLoader::x_ResolveHandle(const CSeq_id_Handle& h)
{
    CGBLGuard g(m_Locks,CGBLGuard::eMain,"x_ResolveHandle");

    CRef<SSeqrefs> sr;
    TSeqId2Seqrefs::iterator bsit = m_Bs2Sr.find(h);
    if (bsit == m_Bs2Sr.end() ) {
        sr.Reset(new SSeqrefs(h));
        m_Bs2Sr[h] = sr;
    }
    else {
        sr = bsit->second;
    }

    int key = sr->m_Handle.GetHash();
    g.Lock(key);
    if( !sr->m_Timer.NeedRefresh(m_Timer) )
        return sr;
    g.Local();

    SSeqrefs::TSeqrefs osr;
    bool got = false;

    CConstRef<CSeq_id> seq_id = h.GetSeqId();
    for ( int try_cnt = 3; !got && try_cnt > 0; --try_cnt ) {
        osr.clear();
        CTimerGuard tg(m_Timer);
        CReader::TConn conn = m_Locks.m_Pool.Select(key);
        try {
            m_Driver->ResolveSeq_id(osr, *seq_id, conn);
            got = true;
            break;
        }
        catch ( CLoaderException& e ) {
            if ( e.GetErrCode() == CLoaderException::eNoConnection ) {
                throw;
            }
            LOG_POST(e.what());
        }
        catch ( const exception &e ) {
            LOG_POST(e.what());
        }
        LOG_POST("GenBank connection failed: Reconnecting....");
        m_Driver->Reconnect(conn);
    }
    if ( !got ) {
        ERR_POST("CGBLoader:x_ResolveHandle: Seq-id resolve failed: "
                 "exceeded maximum attempts count");
        NCBI_THROW(CLoaderException, eLoaderFailed,
                   "Multiple attempts to resolve Seq-id failed");
    }

    g.Lock(); // will unlock everything and lock lookupMutex again 

    swap(sr->m_Sr, osr);
    sr->m_Timer.Reset(m_Timer);
  
    GBLOG_POST( "ResolveHandle(" << h << ") " << sr->m_Sr.size() );
    ITERATE(SSeqrefs::TSeqrefs, srp, sr->m_Sr) {
        GBLOG_POST( (*srp)->print());
    }
    
    if ( !osr.empty() ) {
        bsit = m_Bs2Sr.find(h);
        // make sure we are not deleted in the unlocked time 
        if (bsit != m_Bs2Sr.end()) {
            SSeqrefs::TSeqrefs& nsr=bsit->second->m_Sr;
          
            // catch dissolving TSE and mark them dead
            //GBLOG_POST( "old seqrefs");
            ITERATE ( SSeqrefs::TSeqrefs, srp, osr ) {
                //(*srp)->print(); cout);
                bool found=false;
                ITERATE ( SSeqrefs::TSeqrefs, nsrp, nsr ) {
                    if( (*srp)->SameTSE(**nsrp) ) {
                        found=true;
                        break;
                    }
                }
                if ( found ) {
                    continue;
                }
                TSr2TSEinfo::iterator tsep =
                    m_Sr2TseInfo.find((*srp)->GetKeyByTSE());
                if (tsep == m_Sr2TseInfo.end()) {
                    continue;
                }
              
                // update TSE info 
                CRef<STSEinfo> tse = tsep->second;
                g.Lock(&*tse);
                bool mark_dead  = tse->mode.test(STSEinfo::eDead);
                if ( mark_dead ) {
                    tse->mode.set(STSEinfo::eDead);
                }
                tse->m_SeqIds.erase(h); // drop h as refewrenced seqid
                g.Unlock(&*tse);
                if ( mark_dead && tse->tseinfop ) {
                    // inform data_source :: make sure to avoid deadlocks
                    tse->tseinfop->SetDead(true);
                }
            }
        }
    }
    return sr;
}

bool CGBDataLoader::x_NeedMoreData(const STSEinfo& tse)
{
    bool need_data=true;
  
    if (tse.m_LoadState==STSEinfo::eLoadStateDone)
        need_data=false;
    if (tse.m_LoadState==STSEinfo::eLoadStatePartial) {
        // split code : check tree for presence of data and
        // return from routine if all data already loaded
        // present;
    }
    return need_data;
}


void CGBDataLoader::x_GetData(CRef<STSEinfo> tse,
                              CReader::TConn conn)
{
    try {
        if ( x_NeedMoreData(*tse) ) {
            GBLOG_POST("GetBlob("<<TSE(*tse)<<") "<<":="<<tse->m_LoadState);
            
            _ASSERT(tse->m_LoadState != STSEinfo::eLoadStateDone);
            if (tse->m_LoadState == STSEinfo::eLoadStateNone) {
                CRef<CTSE_Info> tse_info =
                    m_Driver->GetBlob(*tse->seqref, conn);
                GBLOG_POST("GetBlob("<<TSE(*tse)<<")");
                m_InvokeGC=true;
                _ASSERT(tse_info);
                tse->m_LoadState   = STSEinfo::eLoadStateDone;
                tse_info->SetBlobId(tse);
                GetDataSource()->AddTSE(tse_info);
                tse->tseinfop = tse_info.GetPointer();
                _TRACE("GetBlob("<<DUMP(*tse)<<") - whole blob retrieved");
            }
        }
    }
    catch ( CLoaderException& e ) {
        if ( e.GetErrCode() == e.ePrivateData ) {
            LOG_POST("GI("<<tse->seqref->GetGi()<<") is private");
            tse->m_LoadState = STSEinfo::eLoadStateDone;
        }
        else if ( e.GetErrCode() == e.eNoData ) {
            if ( (tse->seqref->GetFlags() & CSeqref::fPossible) ) {
                tse->m_LoadState   = STSEinfo::eLoadStateDone;
            }
            else {
                LOG_POST("ERROR: can not retrive sequence: "<<TSE(*tse));
                throw;
            }
        }
        else {
            throw;
        }
    }
    _ASSERT(!x_NeedMoreData(*tse));
}


void CGBDataLoader::x_GetChunk(CRef<STSEinfo> tse,
                               CReader::TConn conn,
                               CTSE_Chunk_Info& chunk_info)
{
    GBLOG_POST("GetChunk("<<TSE(*tse)<<") "<<":="<<chunk_info.GetChunkId());
  
    _ASSERT(tse->m_LoadState == STSEinfo::eLoadStateDone);
    
    CRef<CTSE_Info> tse_info =
        m_Driver->GetBlob(*tse->seqref, conn, &chunk_info);
    GBLOG_POST("GetChunk("<<TSE(*tse)<<")");
    m_InvokeGC=true;
    _ASSERT(!tse_info);
    _TRACE("GetChunk("<<DUMP(*tse)<<") - chunk retrieved");
}


const CSeqref& CGBDataLoader::GetSeqref(const CTSE_Info& tse_info)
{
    if ( &tse_info.GetDataSource() != GetDataSource() ) {
        NCBI_THROW(CLoaderException, eLoaderFailed,
                   "not mine TSE");
    }
    CRef<CGBDataLoader::STSEinfo> info(GetTSEinfo(tse_info));
    return *info->seqref;
}


void
CGBDataLoader::DebugDump(CDebugDumpContext ddc, unsigned int /*depth*/) const
{
    ddc.SetFrame("CGBLoader");
    // CObject::DebugDump( ddc, depth);
    DebugDumpValue(ddc,"m_TseCount", m_TseCount);
    DebugDumpValue(ddc,"m_TseGC_Threshhold", m_TseGC_Threshhold);
}


END_SCOPE(objects)
END_NCBI_SCOPE



/* ---------------------------------------------------------------------------
* $Log: gbloader.cpp,v $
* Revision 1000.1  2004/06/01 19:41:37  gouriano
* PRODUCTION: UPGRADED [GCC34_MSVC7] Dev-tree R1.106
*
* Revision 1.106  2004/05/21 21:42:52  gorelenk