reader_id1_cache.cpp

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

                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
            if ( CollectStatistics() ) {
                double time = sw.Elapsed();
                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::LoadSNPTable(CSeq_annot_SNP_Info& snp_info,
                                    const CSeqref& seqref)
{
    if ( m_BlobCache ) {
        string key = GetBlobKey(seqref);
        int ver = seqref.GetVersion();
        string subkey = GetSNPTableSubkey();

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

            // table
            CSeq_annot_SNP_Info_Reader::Read(stream, snp_info);

            if ( CollectStatistics() ) {
                double time = sw.Elapsed();
                size_t size = m_BlobCache->GetSize(key, ver, subkey);
                LogBlobStat("CId1Cache: read SNP blob",
                            seqref, size, time);
                snp_load_count++;
                snp_load_bytes += size;
                snp_load_time += time;
            }

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

            if ( CollectStatistics() ) {
                double time = sw.Elapsed();
                LogBlobStat("CId1Cache: read fail SNP blob",
                            seqref, 0, time);
                snp_load_count++;
                snp_load_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;
            }
            CRStream stream(reader.get());

            // blob type
            char type[4];
            if ( !stream.read(type, 4) || memcmp(type, "STBL", 4) != 0 ) {
                if ( CollectStatistics() ) {
                    double time = sw.Elapsed();
                    LogBlobStat("CId1Cache: read fail SNP blob",
                                seqref, 0, time);
                    snp_load_count++;
                    snp_load_time += time;
                }

                return false;
            }

            // table
            CSeq_annot_SNP_Info_Reader::Read(stream, snp_info);

            if ( CollectStatistics() ) {
                double time = sw.Elapsed();
                size_t size = m_OldBlobCache->GetSize(key, ver);
                LogBlobStat("CId1Cache: read SNP blob",
                            seqref, size, time);
                snp_load_count++;
                snp_load_bytes += size;
                snp_load_time += time;
            }

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

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

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


void CCachedId1Reader::StoreSNPTable(const CSeq_annot_SNP_Info& snp_info,
                                     const CSeqref& seqref)
{
    if ( m_BlobCache ) {
        string key = GetBlobKey(seqref);
        int ver = seqref.GetVersion();
        string subkey = GetSNPTableSubkey();

        CStopWatch sw;
        if ( CollectStatistics() ) {
            sw.Start();
        }
        try {
            {{
                auto_ptr<IWriter> writer;
                writer.reset(m_BlobCache->GetWriteStream(key, ver, subkey));
                if ( !writer.get() ) {
                    return;
                }

                {{
                    CWStream stream(writer.get());
                    CSeq_annot_SNP_Info_Reader::Write(stream, snp_info);
                }}
                writer->Flush();
                writer.reset();
            }}

            if ( CollectStatistics() ) {
                double time = sw.Elapsed();
                size_t size = m_BlobCache->GetSize(key, ver, subkey);
                LogBlobStat("CId1Cache: saved SNP blob",
                            seqref, size, time);
                snp_store_count++;
                snp_store_bytes += size;
                snp_store_time += time;
            }
        }
        catch ( exception& exc ) {
            ERR_POST("CId1Cache: "
                     "Exception while storing SNP table: "<<
                     seqref.printTSE() << ": " << exc.what());
            try {
                m_BlobCache->Remove(key);
            }
            catch ( exception& /*exc*/ ) {
                // ignored
            }

            if ( CollectStatistics() ) {
                double time = sw.Elapsed();
                LogBlobStat("CId1Cache: save fail SNP blob",
                            seqref, 0, time);
                snp_store_count++;
                snp_store_time += time;
            }
        }
    }
    else if ( m_OldBlobCache ) {
        string key = GetBlobKey(seqref);
        int ver = seqref.GetVersion();

        CStopWatch sw;
        if ( CollectStatistics() ) {
            sw.Start();
        }
        try {
            {{
                auto_ptr<IWriter> writer;
                writer.reset(m_OldBlobCache->GetWriteStream(key, ver));
                if ( !writer.get() ) {
                    return;
                }

                {{
                    CWStream stream(writer.get());
                    stream.write("STBL", 4);
                    CSeq_annot_SNP_Info_Reader::Write(stream, snp_info);
                }}
                writer->Flush();
                writer.reset();
            }}

            if ( CollectStatistics() ) {
                double time = sw.Elapsed();
                size_t size = m_OldBlobCache->GetSize(key, ver);
                LogBlobStat("CId1Cache: saved SNP blob",
                            seqref, size, time);
                snp_store_count++;
                snp_store_bytes += size;
                snp_store_time += time;
            }
        }        
        catch ( exception& exc ) {
            ERR_POST("CId1Cache: "
                     "Exception while storing SNP table: "<<
                     seqref.printTSE() << ": " << exc.what());
            try {
                m_OldBlobCache->Remove(key);
            }
            catch ( exception& /*exc*/ ) {
                // ignored
            }

            if ( CollectStatistics() ) {
                double time = sw.Elapsed();
                LogBlobStat("CId1Cache: save fail SNP blob",
                            seqref, 0, time);
                snp_store_count++;
                snp_store_time += time;
            }
        }
    }
}


bool CCachedId1Reader::LoadData(const string& key, int version,
                                const char* suffix,
                                CID2_Reply_Data& data)
{
    AutoPtr<IReader> reader(m_BlobCache->GetReadStream(key, version, suffix));
    if ( !reader.get() ) {
        return false;
    }
    
    CIRByteSourceReader rd(reader.get());
    
    CObjectIStreamAsnBinary in(rd);
    
    in >> data;

    return true;
}


bool CCachedId1Reader::LoadData(const string& key, const char* suffix,
                                int version, CID2_Reply_Data& data)
{
    AutoPtr<IReader> reader(m_OldBlobCache->GetReadStream(key + suffix,
                                                          version));
    if ( !reader.get() ) {
        return false;
    }
    
    CIRByteSourceReader rd(reader.get());
    
    CObjectIStreamAsnBinary in(rd);
    
    in >> data;

    return true;
}


class CVectorListReader : public CByteSourceReader
{
public:
    typedef list< vector< char >* > TData;

    CVectorListReader(const TData& data)
        : m_Data(data),
          m_CurrentIter(data.begin()),
          m_CurrentOffset(0)
        {
        }

    size_t Read(char* buffer, size_t bufferLength)
        {
            while ( m_CurrentIter != m_Data.end() ) {
                const vector<char> curr = **m_CurrentIter;
                if ( m_CurrentOffset < curr.size() ) {
                    size_t remaining = curr.size() - m_CurrentOffset;
                    size_t count = min(bufferLength, remaining);
                    memcpy(buffer, &curr[m_CurrentOffset], count);
                    m_CurrentOffset += count;
                    return count;
                }
                ++m_CurrentIter;
                m_CurrentOffset = 0;
            }
            return 0;
        }
    
private:
    const TData& m_Data;
    TData::const_iterator m_CurrentIter;
    size_t m_CurrentOffset;
};


CRef<CByteSourceReader> CCachedId1Reader::GetReader(CID2_Reply_Data& data,
                                                    EDataType data_type)
{
    CRef<CByteSourceReader> ret;
    if ( data.GetData_type() != data_type ) {
        return ret;
    }
    ret.Reset(new CVectorListReader(data.GetData()));
    switch ( data.GetData_compression() ) {
    case eCompression_none:
        break;
    case eCompression_nlm_zip:
        ret.Reset(new CNlmZipBtRdr(ret.GetPointer()));
        break;
    default:
        NCBI_THROW(CLoaderException, eLoaderFailed,
                   "unknown compression");
    }
    return ret;
}


AutoPtr<CObjectIStream> CCachedId1Reader::OpenData(CID2_Reply_Data& data,
                                                   CByteSourceReader& reader)
{
    if ( data.GetData_format() != eSerial_AsnBinary ) {
        NCBI_THROW(CLoaderException, eLoaderFailed,
                   "unknown serial format");
    }
    return new CObjectIStreamAsnBinary(reader);
}


END_SCOPE(objects)
END_NCBI_SCOPE


/*
 * $Log: reader_id1_cache.cpp,v $
 * Revision 1000.2  2004/06/01 19:41:59  gouriano
 * PRODUCTION: UPGRADED [GCC34_MSVC7] Dev-tree R1.25
 *
 * Revision 1.25  2004/05/21 21:42:52  gorelenk
 * Added PCH ncbi_pch.hpp
 *
 * Revision 1.24  2004/04/28 17:06:25  vasilche
 * Load split blobs from new ICache.
 *
 * Revision 1.23  2004/02/17 21:20:11  vasilche
 * Fixed 'unused argument' warning.
 *
 * Revision 1.22  2004/01/22 20:53:31  vasilche
 * Fixed include path.
 *
 * Revision 1.21  2004/01/22 20:10:37  vasilche
 * 1. Splitted ID2 specs to two parts.
 * ID2 now specifies only protocol.
 * Specification of ID2 split data is moved to seqsplit ASN module.
 * For now they are still reside in one resulting library as before - libid2.
 * As the result split specific headers are now in objects/seqsplit.
 * 2. Moved ID2 and ID1 specific code out of object manager.
 * Protocol is processed by corresponding readers.
 * ID2 split parsing is processed by ncbi_xreader library - used by all readers.
 * 3. Updated OBJMGR_LIBS correspondingly.
 *
 * Revision 1.20  2004/01/20 16:56:04  vasilche
 * Allow storing version of any blob (not only SNP).
 *
 * Revision 1.19  2004/01/13 21:54:50  vasilche
 * Requrrected new version
 *
 * Revision 1.5  2004/01/13 16:55:56  vasilche
 * CReader, CSeqref and some more classes moved from xobjmgr to separate lib.
 * Headers moved from include/objmgr to include/objtools/data_loaders/genbank.
 *
 * Revision 1.4  2003/12/30 22:14:42  vasilche
 * Updated genbank loader and readers plugins.
 *
 * Revision 1.17  2003/12/30 16:00:25  vasilche
 * Added support for new ICache (CBDB_Cache) interface.
 *
 * Revision 1.16  2003/12/09 17:30:32  ucko
 * +<stdio.h> for sprintf
 *
 * Revision 1.15  2003/11/26 17:55:59  vasilche
 * Implemented ID2 split in ID1 cache.
 * Fixed loading of splitted annotations.
 *
 * Revision 1.14  2003/10/27 15:05:41  vasilche
 * Added correct recovery of cached ID1 loader if gi->sat/satkey cache is invalid.
 * Added recognition of ID1 error codes: private, etc.
 * Some formatting of old code.
 *
 * Revision 1.13  2003/10/24 15:36:46  vasilche
 * Fixed incorrect order of objects' destruction - IWriter before object stream.
 *
 * Revision 1.12  2003/10/24 13:27:40  vasilche
 * Cached ID1 reader made more safe. Process errors and exceptions correctly.
 * Cleaned statistics printing methods.
 *
 * Revision 1.11  2003/10/23 13:48:38  vasilche
 * Use CRStream and CWStream instead of strstreams.
 *
 * Revision 1.10  2003/10/21 16:32:50  vasilche
 * Cleaned ID1 statistics messages.
 * Now by setting GENBANK_ID1_STATS=1 CId1Reader collects and displays stats.
 * And by setting GENBANK_ID1_STATS=2 CId1Reader logs all activities.
 *
 * Revision 1.9  2003/10/21 14:27:35  vasilche
 * Added caching of gi -> sat,satkey,version resolution.
 * SNP blobs are stored in cache in preprocessed format (platform dependent).
 * Limit number of connections to GenBank servers.
 * Added collection of ID1 loader statistics.
 *
 * Revision 1.8  2003/10/14 21:06:25  vasilche
 * Fixed compression statistics.
 * Disabled caching of SNP blobs.
 *
 * Revision 1.7  2003/10/14 19:31:18  kuznets
 * Removed unnecessary hook in SNP deserialization.
 *
 * Revision 1.6  2003/10/14 18:31:55  vasilche
 * Added caching support for SNP blobs.
 * Added statistics collection of ID1 connection.
 *
 * Revision 1.5  2003/10/08 18:58:23  kuznets
 * Implemented correct ID1 BLOB versions
 *
 * Revision 1.4  2003/10/03 17:41:44  kuznets
 * Added an option, that cache is owned by the ID1 reader.
 *
 * Revision 1.3  2003/10/02 19:29:14  kuznets
 * First working revision
 *
 * Revision 1.2  2003/10/01 19:32:22  kuznets
 * Work in progress
 *
 * Revision 1.1  2003/09/30 19:38:26  vasilche
 * Added support for cached id1 reader.
 *
 */