data_crawler_cache.cpp

Go to the documentation of this file.

Go to the SVN repository for this file.

/*  $Id: data_crawler_cache.cpp 46707 2021-09-09 21:01:01Z rudnev $
 * ===========================================================================
 *
 *                            PUBLIC DOMAIN NOTICE
 *               National Center for Biotechnology Information
 *
 *  This software/database is a "United States Government Work" under the
 *  terms of the United States Copyright Act.  It was written as part of
 *  the author's official duties as a United States Government employee and
 *  thus cannot be copyrighted.  This software/database is freely available
 *  to the public for use. The National Library of Medicine and the U.S.
 *  Government have not placed any restriction on its use or reproduction.
 *
 *  Although all reasonable efforts have been taken to ensure the accuracy
 *  and reliability of the software and data, the NLM and the U.S.
 *  Government do not and cannot warrant the performance or results that
 *  may be obtained by using this software or data. The NLM and the U.S.
 *  Government disclaim all warranties, express or implied, including
 *  warranties of performance, merchantability or fitness for any particular
 *  purpose.
 *
 *  Please cite the author in any work or product based on this material.
 *
 * ===========================================================================
 *
 * Authors:  Vladislav Evgeniev
 *
 * File Description:
 *   
 *
 */
 
#include <ncbi_pch.hpp>
#include <gui/widgets/loaders/data_crawler_cache.hpp>
#include <corelib/ncbiapp.hpp>
#include <corelib/ncbireg.hpp>
#include <corelib/rwstream.hpp>
#include <misc/netstorage/netstorage.hpp>
#include <util/md5.hpp>
#include <chrono>
#include <type_traits>
#include <system_error>
 
 
 
BEGIN_NCBI_SCOPE
 
CDataCrawlerCache::ECacheState CDataCrawlerCache::GetState() const
{
    bool blob_present = m_BlobCache.HasBlob(m_BlobKey, "");
    bool sync_present = m_SyncCache.HasBlob(m_SyncKey, "");
    LOG_POST(Info << "Blob: " << m_BlobKey << " : " << blob_present << " Sync: " << sync_present);
    if (blob_present) {
        if (!sync_present)
            return eState_Created;
    }
    if (sync_present)
        return x_ReadState();
 
    return eState_None;
}
 
CDataCrawlerCache::ECacheState CDataCrawlerCache::SetState(ECacheState desired_state)
{
    ECacheState actual_state = GetState();
    switch(desired_state) {
    case eState_None:
    {
        x_StopHeartbeat();
        {
            std::lock_guard<std::mutex> guard(m_SyncCacheMutex);
            m_SyncCache.Remove(m_SyncKey, 0, "");
        }
        actual_state = eState_None;
        break;
    }
    case eState_InProgress:
    {
        if (actual_state != eState_None)
            return actual_state;
        x_WriteState(eState_InProgress);
        x_StartHeartbeat();
        actual_state = eState_InProgress;
        break;
    }
    case eState_NotAccessible:
    {
        x_StopHeartbeat();
        x_WriteState(eState_NotAccessible, 1800);
        actual_state = eState_NotAccessible;
        break;
    }
    case eState_Created:
    {
        x_StopHeartbeat();
        {
            std::lock_guard<std::mutex> guard(m_SyncCacheMutex);
            m_SyncCache.Remove(m_SyncKey, 0, "");
        }
        actual_state = eState_Created;
        break;
    }
    }
    return actual_state;
}
 
size_t CDataCrawlerCache::GetReadSize() const
{
    return (m_BlobCache.GetSize(m_BlobKey, 0, ""));
}
 
unique_ptr<CNcbiIstream> CDataCrawlerCache::GetReadStream()
{
    unique_ptr<IReader> reader(m_BlobCache.GetReadStream(m_BlobKey, 0, ""));
    return unique_ptr<CNcbiIstream>(new CRStream(reader.release(), 0, 0, CRWStreambuf::fOwnReader) );
}
 
unique_ptr<CNcbiOstream> CDataCrawlerCache::GetWriteStream()
{
    unique_ptr<IWriter> writer(m_BlobCache.GetWriteStream(m_BlobKey, 0, ""));
    return unique_ptr<CNcbiOstream>(new CWStream(writer.release(), 0, 0, CRWStreambuf::fOwnWriter) );
}
 
void CDataCrawlerCache::Remove()
{
    m_BlobCache.Remove(m_BlobKey, 0, "");
}
 
void CDataCrawlerCache::x_Initialize()
{
    m_BlobKey = x_GetCacheKey(eCache_Blob);
    m_SyncKey = x_GetCacheKey(eCache_Sync);
    m_BlobCache = x_InitBlobCache();
    m_SyncCache = x_InitSyncCache();
    m_SyncCache.SetFlags(ICache::fBestReliability);
}
 
string CDataCrawlerCache::x_GetCacheKey(ECacheType type)
{
    string key = x_GetCacheKey();
    const size_t max_key { 120 };
    if (!m_isImmutableKey && key.length() > max_key) { // Shorten key due to NC limits
        const size_t flangs_size { 50 };
        string left = key.substr(0, flangs_size);
        string right = key.substr(key.length() - flangs_size - 1);
        CMD5 md5;
        md5.Update(key.c_str(), key.length());
        key = left;
        key += '_';
        key += md5.GetHexSum();
        key += '_';
        key += right;
    }
    switch(type)
    {
        case eCache_Blob :
            key += "_VcfBlob";
            break;
        case eCache_Sync :
            key += "_VcfSync";
            break;
    }
    return key;
}
 
void CDataCrawlerCache::x_StartHeartbeat()
{
    try {
        LOG_POST(Info << "x_StartHeartbeat");
        m_StopHeartbeat = false;
        m_Heartbeat.reset(new thread(&CDataCrawlerCache::x_Heartbeat, this));
    }
    catch(const std::system_error& e) {
        LOG_POST(Error << "Failed to start the cache heartbeat thread: " << e.what());
        NCBI_THROW(CException, eUnknown, "Failed to start the heartbeat thread.");
    }
}
 
void CDataCrawlerCache::x_StopHeartbeat()
{
    if (!m_Heartbeat)
        return;
    LOG_POST(Info << "x_StopHeartbeat");
    try {
        m_StopHeartbeat = true;
        m_Heartbeat->join();
        m_Heartbeat.reset(nullptr);
    }
    catch(const std::system_error& e) {
        LOG_POST(Error << "Failed to join the cache heartbeat thread: " << e.what());
    }
}
 
void CDataCrawlerCache::x_Heartbeat()
{
    try {
        while(!m_StopHeartbeat) {
            this_thread::sleep_for(chrono::seconds(2));
            x_WriteState(eState_InProgress);
        }
    }
    catch(const CException& e) {
        LOG_POST(Error << "x_Heartbeat failed: " << e);
    }
}
 
CNetICacheClient CDataCrawlerCache::x_InitICache(const string &cache, const string &service, const string &db)
{
    const CNcbiRegistry& reg = CNcbiApplication::Instance()->GetConfig();
    string cache_svc = reg.GetString(cache, "service", service);
    string cache_name = reg.GetString(cache, "cache", db);
    LOG_POST(Info << "x_InitICache, service: " << cache_svc << ", cache: " << cache_name);
    return CNetICacheClient(cache_svc, cache_name, "sviewer");
}
 
inline CDataCrawlerCache::ECacheState CDataCrawlerCache::x_ReadState() const
{
    std::lock_guard<std::mutex> guard(m_SyncCacheMutex);
    underlying_type_t< ECacheState > result { eState_None };
    m_SyncCache.Read(m_SyncKey, 0, "", &result, sizeof(result));
    return static_cast< ECacheState >( result );
}
 
inline void CDataCrawlerCache::x_WriteState(ECacheState state, unsigned int ttl)
{
    std::lock_guard<std::mutex> guard(m_SyncCacheMutex);
    underlying_type_t< ECacheState > st { state };
    m_SyncCache.Store(m_SyncKey, 0, "", &st, sizeof(st), ttl);
}
 
const string CDataCrawlerCache::GetCompoundKey() const
{
    CNetStorageObjectLoc::TVersion version(0);
    return CNetStorageObjectLoc::Create(m_BlobCache.GetService().GetServiceName(), m_BlobCache.GetCacheName(), m_BlobKey, "", version);
}
 
CDataCrawlerCache::~CDataCrawlerCache()
{
    x_StopHeartbeat();
}
 
END_NCBI_SCOPE