reader_id1_cache.cpp

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

            resolve_gi_count++;
            resolve_gi_time += time;
        }
    }
}


bool CCachedId1Reader::GetSeqrefs(const CSeq_id& id, TSeqrefs& srs)
{
    if ( m_IdCache ) {
        return GetSeqrefs(GetIdKey(id), srs);
    }
    else {
        return false;
    }
}


void CCachedId1Reader::StoreSeqrefs(const CSeq_id& id, const TSeqrefs& srs)
{
    if ( m_IdCache ) {
        StoreSeqrefs(GetIdKey(id), srs);
    }
}


int CCachedId1Reader::GetBlobVersion(const CSeqref& seqref)
{
    if ( m_IdCache ) {
        int version = 0;
        if ( x_GetIdCache(GetBlobKey(seqref),
                          GetBlobVersionSubkey(),
                          version) ) {
            return version;
        }
    }
    else if ( m_OldIdCache ) {
        CStopWatch sw;
        if ( CollectStatistics() ) {
            sw.Start();
        }

        vector<int> data;
        if ( !m_OldIdCache->Read(seqref.GetSatKey(), seqref.GetSat(), data) ) {
            if ( CollectStatistics() ) {
                double time = sw.Elapsed();
                LogStat("CId1Cache: failed to get blob version",
                        seqref.printTSE(), time);
                resolve_ver_count++;
                resolve_ver_time += time;
            }
            return 0;
        }
    
        _ASSERT(data.size() == 1);

        if ( CollectStatistics() ) {
            double time = sw.Elapsed();
            LogStat("CId1Cache: got blob version", seqref.printTSE(), time);
            resolve_ver_count++;
            resolve_ver_time += time;
        }

        return data[0];
    }
    return 0;
}


void CCachedId1Reader::StoreBlobVersion(const CSeqref& seqref, int version)
{
    if ( m_IdCache ) {
        x_StoreIdCache(GetBlobKey(seqref),
                       GetBlobVersionSubkey(),
                       version);
    }
    else if ( m_OldIdCache ) {
        CStopWatch sw;
        if ( CollectStatistics() ) {
            sw.Start();
        }

        vector<int> data;
        data.push_back(version);

        _ASSERT(data.size() == 1);

        m_OldIdCache->Store(seqref.GetSatKey(), seqref.GetSat(), data);

        if ( CollectStatistics() ) {
            double time = sw.Elapsed();
            LogStat("CId1Cache: saved blob version", seqref.printTSE(), time);
            resolve_ver_count++;
            resolve_ver_time += time;
        }
    }
}


int CCachedId1Reader::ResolveSeq_id_to_gi(const CSeq_id& id, TConn conn)
{
    if ( m_IdCache ) {
        int gi = 0;
        string key = GetIdKey(id);
        string subkey = GetGiSubkey();
        if ( !x_GetIdCache(key, subkey, gi) ) {
            gi = CId1Reader::ResolveSeq_id_to_gi(id, conn);
            x_StoreIdCache(key, subkey, gi);
        }
        return gi;
    }
    else {
        return CId1Reader::ResolveSeq_id_to_gi(id, conn);
    }
}


void CCachedId1Reader::RetrieveSeqrefs(TSeqrefs& srs, int gi, TConn conn)
{
    if ( !GetSeqrefs(gi, srs) ) {
        CId1Reader::RetrieveSeqrefs(srs, gi, conn);
        StoreSeqrefs(gi, srs);
    }
}


void CCachedId1Reader::GetTSEChunk(const CSeqref& seqref,
                                   CTSE_Chunk_Info& chunk_info,
                                   TConn /*conn*/)
{
    if ( m_BlobCache ) {
        CID2_Reply_Data chunk_data;
        string key = GetBlobKey(seqref);
        string subkey = GetChunkSubkey(chunk_info.GetChunkId());
        if ( !LoadData(key, seqref.GetVersion(), subkey.c_str(),
                       chunk_data) ) {
            NCBI_THROW(CLoaderException, eLoaderFailed,
                       "CCachedId1Reader::GetTSEChunk: chunk is missing");
        }
        CRef<CID2S_Chunk> chunk(new CID2S_Chunk);
        size_t size = 0;
        {{
            CRef<CByteSourceReader> reader = GetReader(chunk_data,
                                                       eDataType_Chunk);
            AutoPtr<CObjectIStream> in(OpenData(chunk_data, *reader));
            CReader::SetSeqEntryReadHooks(*in);
            *in >> *chunk;
            size = in->GetStreamOffset();
        }}
        CSplitParser::Load(chunk_info, *chunk);
    
        // everything is fine
    }
    else if ( m_OldBlobCache ) {
        CStopWatch sw;
        if ( CollectStatistics() ) {
            sw.Start();
        }

        CID2_Reply_Data chunk_data;
        string key = GetBlobKey(seqref);
        string suffix = "-chunk-"+NStr::IntToString(chunk_info.GetChunkId());
        if ( !LoadData(key, suffix.c_str(), seqref.GetVersion(),
                       chunk_data) ) {
            NCBI_THROW(CLoaderException, eLoaderFailed,
                       "CCachedId1Reader::GetTSEChunk: chunk is missing");
        }
        CRef<CID2S_Chunk> chunk(new CID2S_Chunk);
        size_t size = 0;
        {{
            CRef<CByteSourceReader> reader = GetReader(chunk_data,
                                                       eDataType_Chunk);
            AutoPtr<CObjectIStream> in(OpenData(chunk_data, *reader));
            CReader::SetSeqEntryReadHooks(*in);
            *in >> *chunk;
            size = in->GetStreamOffset();
        }}
        CSplitParser::Load(chunk_info, *chunk);
    
        // everything is fine
        if ( CollectStatistics() ) {
            double time = sw.Elapsed();
            LogBlobStat("CId1Cache: read chunk", seqref, size, time);
            chunk_blob_count++;
            chunk_bytes += size;
            chunk_time += time;
        }
    }
    else {
    }
}


int CCachedId1Reader::x_GetVersion(const CSeqref& seqref, TConn conn)
{
    int version = GetBlobVersion(seqref);
    if ( version == 0 ) {
        version = CId1Reader::x_GetVersion(seqref, conn);
        _ASSERT(version != 0);
        StoreBlobVersion(seqref, version);
    }
    return version;
}


void CCachedId1Reader::x_GetTSEBlob(CID1server_back& id1_reply,
                                    CRef<CID2S_Split_Info>& split_info,
                                    const CSeqref& seqref,
                                    TConn conn)
{
    // update seqref's version
    GetVersion(seqref, conn);
    if ( !LoadBlob(id1_reply, split_info, seqref) ) {
        // we'll intercept loading deeper and write loaded data on the fly
        CId1Reader::x_GetTSEBlob(id1_reply, split_info, seqref, conn);
    }
}


void CCachedId1Reader::x_ReadTSEBlob(CID1server_back& id1_reply,
                                     const CSeqref& seqref,
                                     CNcbiIstream& stream)
{
    if ( m_BlobCache ) {
        string key = GetBlobKey(seqref);
        int ver = seqref.GetVersion();
        string subkey = GetSeqEntrySubkey();

        try {
            auto_ptr<IWriter> writer(
                m_BlobCache->GetWriteStream(key, ver, subkey));
            
            if ( writer.get() ) {
                {{
                    CWriterByteSourceReader proxy(&stream, writer.get());
                    
                    CObjectIStreamAsnBinary obj_stream(proxy);
                    
                    CStreamDelayBufferGuard guard(obj_stream);
                    
                    CId1Reader::x_ReadTSEBlob(id1_reply, obj_stream);
                }}

                writer->Flush();

                // everything is fine
                return;
            }
        }
        catch ( ... ) {
            // In case of an error we need to remove incomplete BLOB
            // from the cache.
            try {
                m_BlobCache->Remove(key);
            }
            catch ( exception& /*exc*/ ) {
                // ignored
            }

            // continue with exception
            throw;
        }
    }
    else if ( m_OldBlobCache ) {
        string key = GetBlobKey(seqref);
        int version = seqref.GetVersion();

        try {
            auto_ptr<IWriter> writer(m_OldBlobCache->GetWriteStream(key,
                                                                    version));
            
            if ( writer.get() ) {
                {{
                    CWriterByteSourceReader proxy(&stream, writer.get());
                    
                    CObjectIStreamAsnBinary obj_stream(proxy);
                    
                    CStreamDelayBufferGuard guard(obj_stream);
                    
                    CId1Reader::x_ReadTSEBlob(id1_reply, obj_stream);
                }}

                writer->Flush();
                writer.reset();
                // everything is fine
                return;
            }
        }
        catch ( ... ) {
            // In case of an error we need to remove incomplete BLOB
            // from the cache.
            try {
                m_OldBlobCache->Remove(key);
            }
            catch ( exception& /*exc*/ ) {
                // ignored
            }

            // continue with exception
            throw;
        }
    }
    // by deault read from ID1
    CId1Reader::x_ReadTSEBlob(id1_reply, seqref, stream);
}


void CCachedId1Reader::x_GetSNPAnnot(CSeq_annot_SNP_Info& snp_info,
                                     const CSeqref& seqref,
                                     TConn conn)
{
    // update seqref's version
    GetVersion(seqref, conn);
    if ( !LoadSNPTable(snp_info, seqref) ) {
        snp_info.Reset();
        // load SNP table from GenBank
        CId1Reader::x_GetSNPAnnot(snp_info, seqref, conn);

        // and store SNP table in cache
        StoreSNPTable(snp_info, seqref);
    }
}


bool CCachedId1Reader::LoadBlob(CID1server_back& id1_reply,
                                CRef<CID2S_Split_Info>& split_info,
                                const CSeqref& seqref)
{
    return LoadSplitBlob(id1_reply, split_info, seqref) ||
        LoadWholeBlob(id1_reply, seqref);
}


bool CCachedId1Reader::LoadWholeBlob(CID1server_back& id1_reply,
                                     const CSeqref& seqref)
{
    if ( m_BlobCache ) {
        string key = GetBlobKey(seqref);
        int ver = seqref.GetVersion();
        string subkey = GetSeqEntrySubkey();

        CStopWatch sw;
        if ( CollectStatistics() ) {
            sw.Start();
        }
        try {
            auto_ptr<IReader> reader(
                m_BlobCache->GetReadStream(key, ver, subkey));
            if ( !reader.get() ) {
                return false;
            }

            CIRByteSourceReader rd(reader.get());
        
            CObjectIStreamAsnBinary in(rd);
        
            CReader::SetSeqEntryReadHooks(in);
            in >> id1_reply;
        
            // everything is fine
            if ( CollectStatistics() ) {
                double time = sw.Elapsed();
                size_t size = in.GetStreamOffset();
                LogBlobStat("CId1Cache: read blob", seqref, size, time);
                main_blob_count++;
                main_bytes += size;
                main_time += time;
            }

            return true;
        }
        catch ( exception& exc ) {
            ERR_POST("CId1Cache: Exception while loading cached blob: " <<
                     seqref.printTSE() << ": " << exc.what());

            if ( CollectStatistics() ) {
                double time = sw.Elapsed();
                LogBlobStat("CId1Cache: read fail blob",
                            seqref, 0, time);
                main_blob_count++;
                main_time += time;
            }

            return false;
        }
    }
    else if ( m_OldBlobCache ) {
        string key = GetBlobKey(seqref);
        int ver = seqref.GetVersion();

        CStopWatch sw;
        if ( CollectStatistics() ) {
            sw.Start();
        }
        try {
            auto_ptr<IReader> reader(m_OldBlobCache->GetReadStream(key, ver));
            if ( !reader.get() ) {
                return false;
            }

            CIRByteSourceReader rd(reader.get());
        
            CObjectIStreamAsnBinary in(rd);
        
            CReader::SetSeqEntryReadHooks(in);
            in >> id1_reply;
        
            // everything is fine
            if ( CollectStatistics() ) {
                double time = sw.Elapsed();
                size_t size = in.GetStreamOffset();
                LogBlobStat("CId1Cache: read blob", seqref, size, time);
                main_blob_count++;
                main_bytes += size;
                main_time += time;
            }

            return true;
        }
        catch ( exception& exc ) {
            ERR_POST("CId1Cache: Exception while loading cached blob: " <<
                     seqref.printTSE() << ": " << exc.what());

            if ( CollectStatistics() ) {
                double time = sw.Elapsed();
                LogBlobStat("CId1Cache: read fail blob",
                            seqref, 0, time);
                main_blob_count++;
                main_time += time;
            }

            return false;
        }
    }
    else {
        return false;
    }
}


bool CCachedId1Reader::LoadSplitBlob(CID1server_back& id1_reply,
                                     CRef<CID2S_Split_Info>& split_info,
                                     const CSeqref& seqref)
{
    if ( m_BlobCache ) {
        string key = GetBlobKey(seqref);
        int ver = seqref.GetVersion();

        try {
            CID2_Reply_Data main_data, split_data;
            if ( !LoadData(key, ver, GetSkeletonSubkey(), main_data) ||
                 !LoadData(key, ver, GetSplitInfoSubkey(), split_data) ) {
                return false;
            }
            size_t size = 0;
            CRef<CSeq_entry> main(new CSeq_entry);
            {{
                CRef<CByteSourceReader> reader(GetReader(main_data,
                                                         eDataType_MainBlob));
                AutoPtr<CObjectIStream> in(OpenData(main_data, *reader));
                CReader::SetSeqEntryReadHooks(*in);
                *in >> *main;
                size += in->GetStreamOffset();
            }}
            CRef<CID2S_Split_Info> split(new CID2S_Split_Info);
            {{
                CRef<CByteSourceReader> reader(GetReader(split_data,
                                                         eDataType_SplitInfo));
                AutoPtr<CObjectIStream> in(OpenData(split_data, *reader));
                CReader::SetSeqEntryReadHooks(*in);
                *in >> *split;
                size += in->GetStreamOffset();
            }}
        
            id1_reply.SetGotseqentry(*main);
            split_info = split;

            // everything is fine
            return true;
        }
        catch ( exception& exc ) {
            ERR_POST("CId1Cache: Exception while loading cached blob: " <<
                     seqref.printTSE() << ": " << exc.what());
            return false;
        }
        return false;
    }
    else if ( m_OldBlobCache ) {
        string key = GetBlobKey(seqref);
        int ver = seqref.GetVersion();

        CStopWatch sw;
        if ( CollectStatistics() ) {
            sw.Start();
        }
        try {
            CID2_Reply_Data main_data, split_data;
            if ( !LoadData(key, "-main", ver, main_data) ||
                 !LoadData(key, "-split", ver, split_data) ) {
                return false;
            }
            size_t size = 0;
            CRef<CSeq_entry> main(new CSeq_entry);
            {{