BranchServer.h

Go to the documentation of this file.

//-*- Mode: C++ -*-
//
//  DDC originally by Alexey Sokirko
//  Changes and modifications 2011-2020 by Bryan Jurish
//
//  This file is part of DDC.
// 
//  DDC is free software: you can redistribute it and/or modify
//  it under the terms of the GNU Lesser General Public License as published by
//  the Free Software Foundation, either version 3 of the License, or
//  (at your option) any later version.
// 
//  DDC is distributed in the hope that it will be useful,
//  but WITHOUT ANY WARRANTY; without even the implied warranty of
//  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
//  GNU Lesser General Public License for more details.
// 
//  You should have received a copy of the GNU Lesser General Public License
//  along with DDC.  If not, see <http://www.gnu.org/licenses/>.
//
#ifndef DDC_BRANCH_SERVER_H
#define DDC_BRANCH_SERVER_H

#include "../ConcordDmnLib/Server.h"
#include "../ConcordLib/QueryCompiler.h"

//======================================================================
struct CHitToSort 
{
    BYTE        m_HostNo;       
    size_t      m_HitNo;        
    string      m_SortKeyStr;   

    CHitToSort(BYTE HostNo=0)
        : m_HostNo(HostNo)
    {};

    inline bool operator < (const CHitToSort& X) const {
        return m_SortKeyStr < X.m_SortKeyStr;
    };

    inline void steal(CHitToSort &X)
    {
        m_HostNo = X.m_HostNo;
        m_HitNo  = X.m_HitNo;
        m_SortKeyStr.swap(X.m_SortKeyStr);
    };

    inline string debugStr(void) const
    {
        return Format("{HostNo=%d,HitNo=%zd,SortKey=%s}", m_HostNo, m_HitNo, jsonStr(m_SortKeyStr).c_str());
    };
};

//==============================================================================
struct LessByHitNo {
    bool operator() (const CHitToSort &a, const CHitToSort &b) const
    { return a.m_HitNo < b.m_HitNo; };
};


//======================================================================
struct CFirstHitsQueryResult  {
    size_t                      m_AllHitsCount;
    size_t                      m_AllRelevantDocsCount; 
    string                      m_HitsDistributionStr;
    string                      m_RelevantDocsDistributionStr;
    int                         m_InternalError;
    HitSortOrderEnum            m_SortOrder;            //< hit sort order; formerly bool m_bSort
    NetworkErrorsEnum           m_NetworkError;
    string                      m_ErrorStr;
    vector<CHitToSort>          m_Hits;
    vector<size_t>              m_DtrEnds;              //< EndHitNo components scanned from each immediate daughter subcorpus
    vector<SOCKET>              m_Sockets;
    size_t                      m_Offset;               //< logical offset of m_Hits[0] (sum over number of front-trimmed hits for all subcorpora)
    
    ~CFirstHitsQueryResult()
    {
        CloseSockets();  //-- fix mantis bug #718?
    };

    void Clear();

    inline bool HasErrors() const
    {
        return (m_NetworkError != neSuccess )  || (m_InternalError != errNone);
    };

    void CloseSockets();

    void CollectCounts();

    void PruneHits(size_t PruneLimit);

    size_t HostHitsLimit(const vector<size_t>& HostHitIds);

    inline size_t PhysicalOffset(size_t LogicalHitNo) const
    { return LogicalHitNo - m_Offset; };

    inline size_t LogicalOffset(size_t PhysicalHitNo) const
    { return PhysicalHitNo + m_Offset; };

    inline size_t LogicalMax(size_t LogicalHitNo) const
    { return min(LogicalHitNo, LogicalOffset(m_Hits.size())); };

    inline size_t PhysicalMax(size_t LogicalHitNo) const
    { return PhysicalOffset(LogicalMax(LogicalHitNo)); };

    inline size_t LogicalSize(size_t StartHitNo, size_t ResultLimit) const
    { return LogicalMax(StartHitNo+ResultLimit) - LogicalMax(StartHitNo); };

};

//======================================================================
struct CQueryStatistics 
{
    DwordVector m_HitsEnds;
    DWORD       m_FreqCount;
};

//======================================================================

struct CSelectedHosts
{
    typedef vector<DWORD>            VecT;
    typedef map<DWORD, set<string> > MapT;

    VecT m_vec;  
    MapT m_map;  

    CSelectedHosts()
    {};

    ~CSelectedHosts()
    {};

    inline size_t size() const
    { return m_vec.size(); };

    inline void resize(size_t sz)
    { m_vec.resize(sz); };

    void Clear();

    void Insert(DWORD HostId, const string& SubPath);

    inline void Insert(DWORD HostId)
    { Insert(HostId,"*"); };

    void InsertAll(DWORD nDtrs, const string &SubPath);
    
    string DtrPaths(DWORD HostId) const;

    string toString() const;
};


//======================================================================
class CRunQueryData {
public:
    string sQuery;                      
    string sResultFormat;               
    size_t StartHitNo;                  
    size_t ResultLimit;                 
    int TimeOut;                        
    int iInternalError;                 
    int iNetworkError;                  
    size_t EndHitNo;                    
    size_t HitsCount;                   
    size_t RelevantDocsCount;           
    //DwordVector PossibleHosts;          ///< indices of selected immediate subcorpora
    CSelectedHosts SelectedHosts;       
    string HitDistrib;                  
    string DocDistrib;                  
    string HitsKey;                     
    string ResponseBody;                
    string *pErrorStr;                  
    string     ReqPath;                 
    NavHintKey navKey;                  
    NavHint    navHint;                 
    CQueryCompiler *pQCompiler;         
   
public:
    CRunQueryData(const string & _Query="", const string& _ResultFormat="JSON", size_t _Start=0, size_t _Limit=10, int _TimeOut=10, string *_pErrorStr=NULL);

    virtual ~CRunQueryData();

    void Init(const string & _Query="", const string& _ResultFormat="JSON", size_t _Start=0, size_t _Limit=10, int _TimeOut=10, string *_pErrorStr=NULL, const string& _ReqPath="", const NavHint& _Hint=NavHint());

    void Clear();

    inline CQuery* Query() const
    { return pQCompiler->m_pQuery; };

    inline CQueryOptions* QueryOptions() const
    { return Query()->m_Options; };

    inline bool IsPruneQuery() const
    { return QueryOptions()->IsPruneFilter(); };

    void toString(string &Response, size_t WorkerId=0, const DDCVersionT& vCompat=DDCVersion);

};

//======================================================================
typedef map<string,size_t> CHostIdMap;

//======================================================================
class CDDCBranchServer : public CDDCServer
{
public:

public:
    //--------------------------------------------------------------
    // local data
    map<string, CQueryStatistics>       m_QueryToStatis; //-- unused?
    vector<CHost>                       m_Hosts;
    CHostIdMap                          m_HostMap;       
    string                              m_QueryResultString;
    string                              m_ErrorStr;     //-- subcorpus error

    //-- low-level data
    CFirstHitsQueryResult               m_FHResult;   //< first-hits result
    CRunQueryData                       m_QueryData;  //< run-query data
    NavHintCache                        *m_pNavCache; //< navigation hint cache (shared over multiple worker threads)
    size_t                              m_NavCacheStep; //-- minimum offset increment for update of navigation hint cache

    bool m_bHitStringsParallel; 

public:
    //--------------------------------------------------------------
    // Constructors etc.

    CDDCBranchServer(bool master=true);

    virtual ~CDDCBranchServer();

    virtual void Shutdown();

    virtual void SetOptions(const CDDCServerOptions &Options);

public:
    //--------------------------------------------------------------
    // DDCServer API: client threads

    virtual CDDCServer *WorkerClone(void) const
    { return (CDDCServer*)new CDDCBranchServer(false); };

    virtual void WorkerCloneInit(const CDDCServer *Parent, size_t WorkerId);

    virtual void WorkerCloneFree(void);

public:
    //--------------------------------------------------------------
    // CHost API: overrides
    virtual bool Close(SOCKET& rConnectedSocket, bool linger=false, int timeout=5);

    virtual bool CloseDtr(SOCKET& rConnectedSocket, bool linger=false, int timeout=5);

public:
    //--------------------------------------------------------------
    // DDCServer API: request handlers: required

    /* request handler: get_first_hits: required (leaf) */
    virtual void handle__get_first_hits(string &Result, const char *Query, int TimeOut, size_t ResultLimit, const string& ReqPath, const NavHint& Hint);

    /* request handler: get_hit_strings: required (leaf) */
    virtual void handle__get_hit_strings(string &Result, const char *ResultType, size_t StartHitNo, size_t ResultLimit);

    /* request handler: run_query: required (branch) */
    virtual void handle__run_query(string &Result, const char *CorpusName, const char *Query, const char *ResultType, size_t StartHitNo, size_t ResultLimit, int TimeOut, const string& ReqPath, const NavHint& Hint);

    virtual void handle__expand_terms(string &Result, const string& pipeline, const string& terms, int timeout, const string &corpus);

    virtual void handle__status(string &Result, int timeout=10, bool verbose=false);

    virtual void handle__info(string &Result, int timeout=10, bool verbose=false);

    virtual void handle__reload(string &Result, int depth=0);

    virtual void handle__clear_cache(string &Result, int depth=0);

    virtual void handle__nodes(string &Result, int depth=-1);

    virtual void handle__close_socket(string &Result, SOCKET &rConnectedSocket);

public:
    //--------------------------------------------------------------
    // local methods

    void InitHostMap();

    void GetSubcorpusJsonStrings(vector<string>& Results, const string &request, int timeout=10);

    string GetSubcorpusJsonArray(const string &request, int timeout=10);

    void RunDistributed();

    NetworkErrorsEnum RunGetFirstHits();

    void CheckFirstHitsResult(NetworkErrorsEnum networkError=neSuccess);

    NetworkErrorsEnum RunGetHitStrings();

    NetworkErrorsEnum RunDistributed(string Query,
                                     const string& ResultType,
                                     size_t   StartHitNo,
                                     size_t   ResultLimit,
                                     int&     iInternalError,
                                     size_t&  EndHitNo,
                                     size_t&  HitsCount,
                                     int      TimeOut,
                                     string&  ResultString,
                                     size_t&  RelevantDocsCount,
                                     string&  HitDistrib,
                                     string&  DocDistrib,
                                     string&  HitsKey);

    void UpdateNavCache();

    NetworkErrorsEnum GetHitContexts(CFirstHitsQueryResult& FHResult,
                                     const string& ResultType,
                                     DDCFormatTypeEnum ResultTypeE,
                                     CSelectedHosts& SelectedHosts,
                                     int&    InternalError,
                                     int     TimeOut,
                                     string& ResultString
        );

    NetworkErrorsEnum GetHitCounts(CFirstHitsQueryResult& FHResult, DDCFormatTypeEnum ResultTypeE, string& ResultString);

    void GetFirstHitsFromCorpora(const CSelectedHosts& SelectedHosts,
                                 const string  Query,
                                 const size_t  StartHitNo,
                                 const size_t  ResultLimit,
                                 const int     TimeOut,
                                 CFirstHitsQueryResult& Result);

    NetworkErrorsEnum SendGetHitStringsRequest(BYTE SelectedHostNo,
                                               const vector< vector<size_t> >& SortedHits,
                                               const vector<size_t>& FirstOffsets,
                                               const string& ResultType);
    
    NetworkErrorsEnum GetHitStringsFromOneCorpus(CHost& Host,
                                                 const string& ResultType,
                                                 size_t StartHitNo,
                                                 size_t ResultLimit,
                                                 int   TimeOut,
                                                 SOCKET& CorpusSocket,
                                                 int&    InternalError,
                                                 string& Result,
                                                 string& ResultHint);

    bool   ReadSelectedHosts(string& CleanQuery);

    size_t FindHostID(const string& CorpusName) const;

    CHost* FindHost(const string& CorpusName);

    void AlignCorpusPath(const string& UserPath, string& Dtr, string& Rest);
};

typedef CDDCBranchServer CDDCServerListenHost;

#endif /* DDC_BRANCH_SERVER_H */

/*--- emacs style variables ---
 * Local Variables:
 *  mode: C++
 *  c-file-style: "ellemtel"
 *  c-basic-offset: 4
 *  tab-width: 8
 *  indent-tabs-mode: nil
 * End:
 */