message_handler.cpp

Go to the documentation of this file.

Go to the SVN repository for this file.

/*  $Id: message_handler.cpp 86689 2019-06-06 17:41:53Z gouriano $
 * ===========================================================================
 *
 *                            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.
 *
 * ===========================================================================
 *
 * Author: Pavel Ivanov
 *
 */
 
#include "nc_pch.hpp"
 
#include <corelib/ncbireg.hpp>
#include <corelib/ncbifile.hpp>
#include <corelib/request_ctx.hpp>
#include <corelib/ncbi_bswap.hpp>
#include <util/md5.hpp>
 
#include "netcached.hpp"
#include "message_handler.hpp"
#include "netcache_version.hpp"
#include "nc_stat.hpp"
#include "peer_control.hpp"
#include "distribution_conf.hpp"
#include "periodic_sync.hpp"
#include "active_handler.hpp"
#include "nc_storage.hpp"
#include "nc_storage_blob.hpp"
#include "logging.hpp"
 
 
 
BEGIN_NCBI_SCOPE
 
#if 0
#define LOG_CURRENT_FUNCTION SRV_LOG(Warning, "this: " << (void*)this);
#else
#define LOG_CURRENT_FUNCTION
#endif
 
// when 1, server always broadcasts 'blob update' notifications.
// This will create empty blob stub on COPY_UPD if needed
// This takes time though, and I am not sure this is required.
// Another problem is that it overloads communication channels - 
// NC fails much more often in CNCPeerControl::x_ReserveBGConn (too many connections)
#define USE_ALWAYS_COPY_UPD  0
 
/// Definition of all NetCache commands
/// 
/// General format of a "NetCache" command is as follows:
/// 
/// CMD param1 param2 ...
/// 
/// Format of "ICache" command is as follows:
/// 
/// IC(cache) CMD param1 param2 ...
/// 
/// Here "IC" is two letters that appear in command literally. "cache" is name
/// of the cache where blob is stored; it's mentioned in parameter list as
/// first parameter with type eNSPA_ICPrefix. Every command parameter can be
/// given as just value or as name=value pair. String parameter values can be
/// enclosed in double quotes. If parameter is declared with the flag
/// eNSPA_Optional then it can be skipped from the command, in this case its
/// default value will be used (if any). If parameter flag eNSPA_Optchain has
/// the same meaning except if it's not provided then all following parameters
/// marked as eNSPA_Optional will be assumed not provided too.
/// 
/// If command needs some binary data along with it then it's sent split in
/// chunks each having 4-byte integer prefix containing the length of the chunk.
/// When all data is sent special chunk length 0xFFFFFFFF should be sent
/// at the end. Successful response of each command is sent as one line
/// starting with "OK:" and then space-separated parameters that need to be
/// returned. If response should contain binary data then initial response
/// line should have "SIZE=nnn" with the size of binary data to follow.
/// Unsuccessful responses to commands always start with "ERR:" and then error
/// explanation follows.
/// 
/// Descriptions of commands have
///  - command name as it comes from client;
///  - structure containing
///    * state function processing this command;
///    * command name as it appears in statistics;
///    * flags controlling command behavior;
///    * type of access to blob if needed;
///    * type of proxy command that should be executed if command will need to
///      be proxied to other servers;
///  - set of structures explaining command parameters. Each structure has
///    * parameter name which can be used by client if it passes parameters in
///      name=value form. Also name is used to distinguish parameters in
///      x_AssignCmdParams();
///    * type of parameter - parser makes additional checks to see if given
///      value is applicable for necessary parameter type;
///    * parameter flags.
static CNCMessageHandler::SCommandDef s_CommandMap[] = {
    // "Are you alive?" command. This is old and deprecated command but it's
    // executed a lot in old ICache clients. All that they need in response is
    // "OK:".
    { "A?",
        {&CNCMessageHandler::x_FinishCommand,
            "A?", fNoCmdFlags, eNCNone, eProxyNone} },
    // Requests version of the server.
    { "VERSION",
        {&CNCMessageHandler::x_DoCmd_Version,
            "VERSION", fNoCmdFlags, eNCNone, eProxyNone},
        {
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Requests some "health" information about the server.
    { "HEALTH",
        {&CNCMessageHandler::x_DoCmd_Health,
            "HEALTH", fNoCmdFlags, eNCNone, eProxyNone},
        {
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Check if blob exists. Command for "ICache" clients.
    { "HASB",
        {&CNCMessageHandler::x_DoCmd_HasBlob,
            "IC_HASB",
            eClientBlobRead | fPeerFindExistsOnly | fNoBlobVersionCheck,
            eNCRead,
            eProxyHasBlob},
          // Name of cache for blob.
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's version.
          { "version", eNSPT_Int,  eNSPA_Required },
          // Blob's subkey.
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Read blob contents. Command for "ICache" clients.
    { "READ",
        {&CNCMessageHandler::x_DoCmd_Get,
            "IC_READ",
            eClientBlobRead,
            eNCReadData,
            eProxyRead},
          // Name of cache for blob.
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's version.
          { "version", eNSPT_Int,  eNSPA_Required },
          // Blob's subkey.
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // Max age of blob (returned blob should be younger)
          { "age",     eNSPT_Int,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Write blob contents. Command for "ICache" clients.
    { "STOR",
        {&CNCMessageHandler::x_DoCmd_IC_Store,
            "IC_STOR",
            eClientBlobWrite | fNoReplyOnFinish,
            eNCCreate,
            eProxyWrite},
          // Name of cache for blob.
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix },
          // Time-to-live for the blob, 0 means default from server settings.
          { "ttl",     eNSPT_Int,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's version.
          { "version", eNSPT_Int,  eNSPA_Required },
          // Blob's subkey.
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // 1 if client wants confirmation after blob has been written
          // (by default it just assumes that everything is written and moves
          // further).
          { "confirm", eNSPT_Int,  eNSPA_Optional },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional },
          // see ENCUserFlags, added in v6.11.0 (CXX-8737)
          { "flags",  eNSPT_Int,  eNSPA_Optional }
        } },
    // Write blob contents. Old and deprecated command which probably is not
    // used by modern ICache clients anymore. It has the size of the blob right
    // in the command (so client should know it beforehand) and it doesn't use
    // "EOF" marker at the end of blob data.
    { "STRS",
        {&CNCMessageHandler::x_DoCmd_IC_Store,
            "IC_STRS",
            eClientBlobWrite | fReadExactBlobSize | fSkipBlobEOF | fNoReplyOnFinish,
            eNCCreate,
            eProxyWrite},
          // Name of cache for blob.
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix },
          // Time-to-live for the blob, 0 means default from server settings.
          { "ttl",     eNSPT_Int,  eNSPA_Required },
          // Size of the blob to be written.
          { "size",    eNSPT_Int,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's version.
          { "version", eNSPT_Int,  eNSPA_Required },
          // Blob's subkey.
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Read all or part of contents of the "last" version of the blob.
    // In response to the command NC sends blob contents, blob version it has
    // and flag showing if this version can be considered "valid", i.e. if it's
    // not expired yet.
    { "READLAST",
        {&CNCMessageHandler::x_DoCmd_GetLast,
            "IC_READLAST",
            eClientBlobRead | fNoBlobVersionCheck,
            eNCReadData,
            eProxyReadLast},
          // Name of cache for blob.
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey.
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Starting position of the data that needs to be sent.
          { "start",   eNSPT_Int,  eNSPA_Optional },
          // Size of the data that needs to be sent.
          { "size",    eNSPT_Int,  eNSPA_Optional },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // Max age of blob (returned blob should be younger)
          { "age",     eNSPT_Int,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Mark the given blob version as "valid" and do that only if this version
    // is still current and wasn't rewritten with another version.
    { "SETVALID",
        {&CNCMessageHandler::x_DoCmd_SetValid,
            "IC_SETVALID",
            fNeedsBlobAccess,
            eNCRead,
            eProxySetValid},
          // Name of cache for blob.
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's version.
          { "version", eNSPT_Int,  eNSPA_Required },
          // Blob's subkey.
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Write blob contents. Command is issued only by other servers while
    // mirroring just written blobs or processing quorum requirements,
    // i.e. writing to other servers before answering to client that blob is
    // written.
    { "COPY_PUT",
        {&CNCMessageHandler::x_DoCmd_CopyPut,
            "COPY_PUT",
            eCopyBlobFromPeer | fNeedsSpaceAsPeer | fReadExactBlobSize
                              | fCopyLogEvent,
            eNCCopyCreate, eProxyNone},
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Blob's version (for NC-generated blob keys this will be equal to 0).
          { "version", eNSPT_Int,  eNSPA_Required },
          // MD5 checksum of blob's password.
          { "md5_pass",eNSPT_Str,  eNSPA_Required },
          // Creation time of the blob (microseconds since epoch).
          { "cr_time", eNSPT_Int,  eNSPA_Required },
          // Time-to-live for the blob.
          { "ttl",     eNSPT_Int,  eNSPA_Required },
          // Dead-time for the blob (can be greater than expiration time).
          { "dead",    eNSPT_Int,  eNSPA_Required },
          // Expiration time for the blob
          { "exp",     eNSPT_Int,  eNSPA_Required },
          // Blob size.
          { "size",    eNSPT_Int,  eNSPA_Required },
          // Time-to-live for blob's version.
          { "ver_ttl", eNSPT_Int,  eNSPA_Required },
          // Blob's version expiration time.
          { "ver_dead",eNSPT_Int,  eNSPA_Required },
          // Server_id of the server where blob was created.
          { "cr_srv",  eNSPT_Int,  eNSPA_Required },
          // Id of the blob on the server where it was created.
          { "cr_id",   eNSPT_Int,  eNSPA_Required },
          // Record number of the event of blob creation in synchronization
          // logs of the server where blob was created.
          { "log_rec", eNSPT_Int,  eNSPA_Required },
          // Version of the command. Field exists for protocol backwards
          // compatibility with previous versions of NC. In current NC this
          // version is always 1.
          { "cmd_ver", eNSPT_Int,  eNSPA_Optional, "0" },
          // Client IP for application that requested writing the blob.
          // Parameter is not empty only if command is issued as part of
          // quorum-related functionality, i.e. before client received
          // confirmation of blob writing.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application that requested writing the blob.
          // Parameter is not empty only if command is issued as part of
          // quorum-related functionality, i.e. before client received
          // confirmation of blob writing.
          { "sid",     eNSPT_Str,  eNSPA_Optional } } },
    // Prolong blob lifetime. Command is issued only by other servers while
    // mirroring prolonged blobs.
    { "COPY_PROLONG",
        {&CNCMessageHandler::x_DoCmd_CopyProlong,
            "COPY_PROLONG",
            eCopyBlobFromPeer,
            eNCRead, eProxyNone},
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Creation time of the blob (microseconds since epoch).
          { "cr_time", eNSPT_Int,  eNSPA_Required },
          // Server_id of the server where blob was created.
          { "cr_srv",  eNSPT_Int,  eNSPA_Required },
          // Id of the blob on the server where it was created.
          { "cr_id",   eNSPT_Int,  eNSPA_Required },
          // Dead-time for the blob (can be greater than expiration time).
          { "dead",    eNSPT_Int,  eNSPA_Required },
          // Expiration time for the blob
          { "exp",     eNSPT_Int,  eNSPA_Required },
          // Blob's version expiration time.
          { "ver_dead",eNSPT_Int,  eNSPA_Required },
          // Time of creation of initial record in synchronization log about
          // this operation.
          { "log_time",eNSPT_Int,  fNSPA_Optional },
          // Server that first made the blob's life prolongation.
          { "log_srv", eNSPT_Int,  eNSPA_Optional },
          // Record number of the initial record in synchronization log about
          // this operation.
          { "log_rec", eNSPT_Int,  eNSPA_Optional } } },
    // Write blob contents. Command for "NetCache" clients.
    { "PUT3",
        {&CNCMessageHandler::x_DoCmd_Put,
            "PUT3",
            eClientBlobWrite | fCanGenerateKey,
            eNCCreate,
            eProxyWrite},
          // Time-to-live for the blob. If not given or 0 then default TTL
          // is used.
        { { "ttl",     eNSPT_Int,  eNSPA_Optional },
          // Key of the blob. If it's not given or empty then new key will be
          // generated.
          { "key",     eNSPT_NCID, eNSPA_Optional },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional },
          // see ENCUserFlags, added in v6.11.0 (CXX-8737)
          { "flags",  eNSPT_Int,  eNSPA_Optional }
        } },
    // Read blob contents. Command for "NetCache" clients.
    { "GET2",
        {&CNCMessageHandler::x_DoCmd_Get,
            "GET2",
            eClientBlobRead,
            eNCReadData,
            eProxyRead},
          // Key of the blob.
        { { "key",     eNSPT_NCID, eNSPA_Required },
          // Not used and not implemented parameter. Exists just for backwards
          // compatibility with old clients.
          { "NW",      eNSPT_Id,   eNSPA_Obsolete | fNSPA_Match },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // Max age of blob (returned blob should be younger)
          { "age",     eNSPT_Int,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Check if blob exists. Command for "NetCache" clients.
    { "HASB",
        {&CNCMessageHandler::x_DoCmd_HasBlob,
            "HASB",
            eClientBlobRead | fPeerFindExistsOnly,
            eNCRead,
            eProxyHasBlob},
          // Key of the blob.
        { { "key",     eNSPT_NCID, eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Delete blob. Command for "NetCache" clients.
    // If the blob doesn't exist command is still considered successful.
    { "RMV2",
        {&CNCMessageHandler::x_DoCmd_Remove,
            "RMV2",
            fNeedsBlobAccess | fNoBlobAccessStats,
            eNCCreate,
            eProxyRemove},
          // Key of the blob.
        { { "key",     eNSPT_NCID, eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Get size of the blob. Command for "NetCache" clients.
    { "GSIZ",
        {&CNCMessageHandler::x_DoCmd_GetSize,
            "GetSIZe",
            eClientBlobRead,
            eNCRead,
            eProxyGetSize},
          // Key of the blob.
        { { "key",     eNSPT_NCID, eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Delete blob. Command for "ICache" clients.
    // If the blob doesn't exist command is still considered successful.
    { "REMO",
        {&CNCMessageHandler::x_DoCmd_Remove,
            "IC_REMOve",
            fNeedsBlobAccess | fNoBlobAccessStats,
            eNCCreate,
            eProxyRemove},
          // Name of cache for blob.
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's version.
          { "version", eNSPT_Int,  eNSPA_Required },
          // Blob's subkey.
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Get size of the blob. Command for "ICache" clients.
    { "GSIZ",
        {&CNCMessageHandler::x_DoCmd_GetSize,
            "IC_GetSIZe",
            eClientBlobRead,
            eNCRead,
            eProxyGetSize},
          // Name of cache for blob.
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's version.
          { "version", eNSPT_Int,  eNSPA_Required },
          // Blob's subkey.
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Read part of the blob contents. Command for "NetCache" clients.
    { "GETPART",
        {&CNCMessageHandler::x_DoCmd_Get,
            "GETPART",
            eClientBlobRead,
            eNCReadData,
            eProxyRead},
          // Key of the blob.
        { { "key",     eNSPT_NCID, eNSPA_Required },
          // Starting position of the data that needs to be sent.
          { "start",   eNSPT_Int,  eNSPA_Required },
          // Size of the data that needs to be sent.
          { "size",    eNSPT_Int,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // Max age of blob (returned blob should be younger)
          { "age",     eNSPT_Int,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Read part of the blob contents. Command for "ICache" clients.
    { "READPART",
        {&CNCMessageHandler::x_DoCmd_Get,
            "IC_READPART",
            eClientBlobRead,
            eNCReadData,
            eProxyRead},
          // Name of cache for blob.
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's version.
          { "version", eNSPT_Int,  eNSPA_Required },
          // Blob's subkey.
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Starting position of the data that needs to be sent.
          { "start",   eNSPT_Int,  eNSPA_Required },
          // Size of the data that needs to be sent.
          { "size",    eNSPT_Int,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // Max age of blob (returned blob should be younger)
          { "age",     eNSPT_Int,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Get meta information about the blob. This command is sent only by other
    // NC servers. And it's used to determine which server has the latest
    // version of the blob. Thus response to this command is a line containing
    // enough information to compare the blob's creation time with the same
    // blob on other servers.
    { "PROXY_META",
        {&CNCMessageHandler::x_DoCmd_ProxyMeta,
            "PROXY_META",
            fNeedsBlobAccess | fNeedsStorageCache | fDoNotProxyToPeers
                             | fDoNotCheckPassword,
            eNCRead, eProxyNone},
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Client IP for application on behalf of which the info is requested.
          { "ip",      eNSPT_Str,  eNSPA_Required },
          // Session ID for application on behalf of which the info is requested.
          { "sid",     eNSPT_Str,  eNSPA_Required } } },
 
    // another NC server notifies this one that the blob was updated there
    // the blob data will come later (eg, via COPY_PUT)
    { "COPY_UPD",
        {&CNCMessageHandler::x_DoCmd_CopyUpdate,
            "UPD",
            fNoBlobAccessStats  | 
            fNeedsBlobAccess | fNeedsStorageCache  | fDoNotProxyToPeers
                             | fDoNotCheckPassword | fNoBlobVersionCheck,
#if USE_ALWAYS_COPY_UPD
            eNCCopyCreate
#else
            eNCRead
#endif
             , eProxyNone },
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // time of the update
          { "cr_time", eNSPT_Int,  eNSPA_Required },
          // Server_id, where the update was made 
          { "cr_srv",  eNSPT_Int,  eNSPA_Required }
        }
    },
    // Write the blob contents. This command is sent only by other NC servers
    // in response to client's PUT3 (or similar) command when the server where
    // client have sent initial command cannot execute it locally for any
    // reason (slot is not processed by that server or initial database caching
    // is not completed yet).
    { "PROXY_PUT",
        {&CNCMessageHandler::x_DoCmd_Put,
            "PROXY_PUT",
            eProxyBlobWrite | fDoNotProxyToPeers,
            eNCCreate,
            eProxyWrite},
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Blob's version (for NC-generated blob keys this will be equal to 0).
          { "version", eNSPT_Int,  eNSPA_Required },
          // Time-to-live for the blob.
          { "ttl",     eNSPT_Int,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Required },
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  eNSPA_Required },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Required },
          // see ENCUserFlags, added in v6.11.0 (CXX-8737)
          { "flags",  eNSPT_Int,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional }
        } },
    // Read all or a part of the blob contents. This command is sent only
    // by other NC servers in response to client's GET2 (or similar) command
    // when the server where client have sent initial command cannot execute it
    // locally for any reason (slot is not processed by that server, or initial
    // database caching is not completed yet, or it was determined that this
    // server has latest version of the blob).
    { "PROXY_GET",
        {&CNCMessageHandler::x_DoCmd_Get,
            "PROXY_GET",
            eProxyBlobRead | fDoNotProxyToPeers,
            eNCReadData,
            eProxyRead},
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Blob's version (for NC-generated blob keys this will be equal to 0).
          { "version", eNSPT_Int,  eNSPA_Required },
          // Starting position of the data that needs to be sent.
          { "start",   eNSPT_Int,  eNSPA_Required },
          // Size of the data that needs to be sent.
          { "size",    eNSPT_Int,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Required },
          // Value of the flag "search_on_read" to use with this command, i.e.
          // whether this server should search the blob if it's not found
          // locally.
          { "srch",    eNSPT_Int,  eNSPA_Required },
          // Flag whether local execution of this command should be forced
          // no matter what, i.e. if this flag is set in cases when normal GET2
          // command would have been proxied to other servers an error should
          // be returned.
          { "local",   eNSPT_Int,  eNSPA_Required },
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  eNSPA_Required },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Required },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // Max age of blob (returned blob should be younger)
          { "age",     eNSPT_Int,  eNSPA_Optional }
        } },
    // Check if the blob exists. This command is sent only by other NC servers
    // in response to client's HASB command when the server where
    // client have sent initial command cannot execute it locally for any
    // reason (slot is not processed by that server or initial database caching
    // is not completed yet).
    { "PROXY_HASB",
        {&CNCMessageHandler::x_DoCmd_HasBlob,
            "PROXY_HASB",
            eProxyBlobRead | fPeerFindExistsOnly | fDoNotProxyToPeers,
            eNCRead,
            eProxyHasBlob},
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Required },
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  eNSPA_Required },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Required },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional } } },
    // Get size of the blob. This command is sent only
    // by other NC servers in response to client's GSIZ command
    // when the server where client have sent initial command cannot execute it
    // locally for any reason (slot is not processed by that server, or initial
    // database caching is not completed yet, or it was determined that this
    // server has latest version of the blob).
    { "PROXY_GSIZ",
        {&CNCMessageHandler::x_DoCmd_GetSize,
            "PROXY_GetSIZe",
            eProxyBlobRead | fDoNotProxyToPeers,
            eNCRead,
            eProxyGetSize},
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Blob's version (for NC-generated blob keys this will be equal to 0).
          { "version", eNSPT_Int,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Required },
          // Value of the flag "search_on_read" to use with this command, i.e.
          // whether this server should search the blob if it's not found
          // locally.
          { "srch",    eNSPT_Int,  eNSPA_Required },
          // Flag whether local execution of this command should be forced
          // no matter what, i.e. if this flag is set in cases when normal GET2
          // command would have been proxied to other servers an error should
          // be returned.
          { "local",   eNSPT_Int,  eNSPA_Required },
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  eNSPA_Required },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Required },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional } } },
    // Read all or a part of contents of the "last version" of the blob.
    // This command is sent only by other NC servers in response to client's
    // READLAST command when the server where client have sent initial command
    // cannot execute it locally for any reason (slot is not processed by that
    // server, or initial database caching is not completed yet, or it was
    // determined that this server has latest version of the blob).
    { "PROXY_READLAST",
        {&CNCMessageHandler::x_DoCmd_GetLast,
            "PROXY_READLAST",
            eProxyBlobRead | fNoBlobVersionCheck | fDoNotProxyToPeers,
            eNCReadData,
            eProxyReadLast},
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Starting position of the data that needs to be sent.
          { "start",   eNSPT_Int,  eNSPA_Required },
          // Size of the data that needs to be sent.
          { "size",    eNSPT_Int,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Required },
          // Value of the flag "search_on_read" to use with this command, i.e.
          // whether this server should search the blob if it's not found
          // locally.
          { "srch",    eNSPT_Int,  eNSPA_Required },
          // Flag whether local execution of this command should be forced
          // no matter what, i.e. if this flag is set in cases when normal
          // READLAST command would have been proxied to other servers an error
          // should be returned.
          { "local",   eNSPT_Int,  eNSPA_Required },
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  eNSPA_Required },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Required },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // Max age of blob (returned blob should be younger)
          { "age",     eNSPT_Int,  eNSPA_Optional } } },
    // Mark the "current version" of the blob as "valid".
    // This command is sent only by other NC servers in response to client's
    // SETVALID command when the server where client have sent initial command
    // cannot execute it locally for any reason (slot is not processed by that
    // server, or initial database caching is not completed yet).
    { "PROXY_SETVALID",
        {&CNCMessageHandler::x_DoCmd_SetValid,
            "PROXY_SETVALID",
            fNeedsBlobAccess | fNeedsStorageCache | fDoNotProxyToPeers,
            eNCRead,
            eProxySetValid},
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Blob's version (for NC-generated blob keys this will be equal to 0).
          { "version", eNSPT_Int,  eNSPA_Required },
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  eNSPA_Required },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Required },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional } } },
    // Remove the blob. This command is sent only by other NC servers in response
    // to client's RMV2 (or similar) command when the server where client have
    // sent initial command cannot execute it locally for any reason (slot is not
    // processed by that server, or initial database caching is not completed yet).
    { "PROXY_RMV",
        {&CNCMessageHandler::x_DoCmd_Remove,
            "PROXY_ReMoVe",
            fNeedsBlobAccess | fNoBlobAccessStats | fDoNotProxyToPeers,
            eNCCreate,
            eProxyRemove},
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Blob's version (for NC-generated blob keys this will be equal to 0).
          { "version", eNSPT_Int,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Required },
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  eNSPA_Required },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Required },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional }
        } },
    // Remove the blob. This command is sent by other NC servers in certain scenarios only
    { "COPY_RMV",
        {&CNCMessageHandler::x_DoCmd_Remove,
            "COPY_ReMoVe",
            fNeedsBlobAccess | fNoBlobAccessStats | fDoNotProxyToPeers
                             | fDoNotCheckPassword | fNoBlobVersionCheck | fCopyLogEvent,
            eNCRead,
            eProxyRemove},
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // time of the update
          { "cr_time", eNSPT_Int, eNSPA_Required },
          // Server_id, requestor
          { "cr_srv",  eNSPT_Int,  eNSPA_Required }
        }
    },
    // Read meta information about the blob. This command is sent only by other
    // NC servers in response to client's GETMETA command when the server where
    // client have sent initial command cannot execute it locally for any reason
    // (slot is not processed by that server, or initial database caching
    // is not completed yet, or it was determined that this server has latest
    // version of the blob).
    { "PROXY_GETMETA",
        {&CNCMessageHandler::x_DoCmd_GetMeta,
            "PROXY_GETMETA",
            eProxyBlobRead | fDoNotCheckPassword | fDoNotProxyToPeers,
            eNCRead,
            eProxyGetMeta},
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Required },
          // Flag whether local execution of this command should be forced
          // no matter what, i.e. if this flag is set in cases when normal
          // GETMETA command would have been proxied to other servers an error
          // should be returned.
          { "local",   eNSPT_Int,  eNSPA_Required },
          // Client IP for application requesting the info.
          { "ip",      eNSPT_Str,  eNSPA_Required },
          // Session ID for application requesting the info.
          { "sid",     eNSPT_Str,  eNSPA_Required },
          // HTTP flag: read input in NC format, write output in HTTP one
          { "http", eNSPT_Int,  eNSPA_Optional }
        } },
    // Start periodic synchronization session. Command is sent only by other
    // NC servers when CNCActiveSyncControl in them decides to start
    // synchronization. Response to this command contains list of events from
    // sync logs of this server which need to be synchronized. Or if this
    // server understands that synchronization using blob lists is needed then
    // first line of response will contain ALL_BLOBS word and then full list
    // of blobs in this slot will be sent.
    { "SYNC_START",
        {&CNCMessageHandler::x_DoCmd_SyncStart,
            "SYNC_START",
          // this command does not need space, but, to do sync, we WILL need space
            fNeedsSpaceAsPeer |
            fNeedsStorageCache | fNeedsLowerPriority | fNeedsAdminClient
            , eNCNone, eProxyNone},
          // Server id of the server starting synchronization.
        { { "srv_id",  eNSPT_Int,  eNSPA_Required },
          // Slot to start synchronization on.
          { "slot",    eNSPT_Int,  eNSPA_Required },
          // Last synchronized record number (in sync log) of _that_ server
          // as _that_ server thinks.
          { "rec_my",  eNSPT_Int,  eNSPA_Required },
          // Last synchronized record number (in sync log) of _this_ server
          // as _that_ server thinks.
          { "rec_your",eNSPT_Int,  eNSPA_Required } } },
    // Get full list of blobs for the slot. Command is sent only by other NC
    // servers when that server decides that synchronization using blob lists
    // is needed. Command can be sent only after successful execution of
    // SYNC_START command.
    { "SYNC_BLIST",
        {&CNCMessageHandler::x_DoCmd_SyncBlobsList,
            "SYNC_BLIST",
            eRunsInStartedSync, eNCNone, eProxyNone},
          // Server id of the server managing the synchronization.
        { { "srv_id",  eNSPT_Int,  eNSPA_Required },
          // Slot that synchronization is started on.
          { "slot",    eNSPT_Int,  eNSPA_Required } } },
    // Write blob contents. This command is sent only by other NC servers
    // during synchronization session if some blob was written on that server
    // and the same data didn't make it to this server yet.
    { "SYNC_PUT",
        {&CNCMessageHandler::x_DoCmd_CopyPut,
            "SYNC_PUT",
            eSyncBlobCmd | fNeedsSpaceAsPeer
                         | fReadExactBlobSize | fCopyLogEvent,
            eNCCopyCreate, eProxyNone},
          // Server id of the server managing the synchronization.
        { { "srv_id",  eNSPT_Int,  eNSPA_Required },
          // Slot that synchronization is started on.
          { "slot",    eNSPT_Int,  eNSPA_Required },
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
          { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Blob's version (for NC-generated blob keys this will be equal to 0).
          { "version", eNSPT_Int,  eNSPA_Required },
          // MD5 checksum of blob's password.
          { "md5_pass",eNSPT_Str,  eNSPA_Required },
          // Creation time of the blob (microseconds since epoch).
          { "cr_time", eNSPT_Int,  eNSPA_Required },
          // Time-to-live for the blob.
          { "ttl",     eNSPT_Int,  eNSPA_Required },
          // Dead-time for the blob (can be greater than expiration time).
          { "dead",    eNSPT_Int,  eNSPA_Required },
          // Expiration time for the blob
          { "exp",     eNSPT_Int,  eNSPA_Required },
          // Blob size.
          { "size",    eNSPT_Int,  eNSPA_Required },
          // Time-to-live for blob's version.
          { "ver_ttl", eNSPT_Int,  eNSPA_Required },
          // Blob's version expiration time.
          { "ver_dead",eNSPT_Int,  eNSPA_Required },
          // Server_id of the server where blob was created.
          { "cr_srv",  eNSPT_Int,  eNSPA_Required },
          // Id of the blob on the server where it was created.
          { "cr_id",   eNSPT_Int,  eNSPA_Required },
          // Record number of the event of blob creation in synchronization
          // logs of the server where blob was created.
          { "log_rec", eNSPT_Int,  eNSPA_Required },
          // Version of the command. Field exists for protocol backwards
          // compatibility with previous versions of NC. In current NC this
          // version is always 1.
          { "cmd_ver", eNSPT_Int,  eNSPA_Optional, "0" } } },
    // Prolong the blob's life. This command is sent only by other NC servers
    // during synchronization session if some blob was prolonged on that server
    // and the same prolongation didn't happen on this server yet.
    { "SYNC_PROLONG",
        {&CNCMessageHandler::x_DoCmd_CopyProlong,
            "SYNC_PROLONG",
            eSyncBlobCmd,
            eNCRead, eProxyNone},
          // Server id of the server managing the synchronization.
        { { "srv_id",  eNSPT_Int,  eNSPA_Required },
          // Slot that synchronization is started on.
          { "slot",    eNSPT_Int,  eNSPA_Required },
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
          { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Creation time of the blob (microseconds since epoch).
          { "cr_time", eNSPT_Int,  eNSPA_Required },
          // Server_id of the server where blob was created.
          { "cr_srv",  eNSPT_Int,  eNSPA_Required },
          // Id of the blob on the server where it was created.
          { "cr_id",   eNSPT_Int,  eNSPA_Required },
          // Dead-time for the blob (can be greater than expiration time).
          { "dead",    eNSPT_Int,  eNSPA_Required },
          // Expiration time for the blob
          { "exp",     eNSPT_Int,  eNSPA_Required },
          // Blob's version expiration time.
          { "ver_dead",eNSPT_Int,  eNSPA_Required },
          // Time of creation of initial record in synchronization log about
          // this operation.
          { "log_time",eNSPT_Int,  fNSPA_Optional },
          // Server that first made the blob's life prolongation.
          { "log_srv", eNSPT_Int,  eNSPA_Optional },
          // Record number of the initial record in synchronization log about
          // this operation.
          { "log_rec", eNSPT_Int,  eNSPA_Optional } } },
    // Read blob contents. This command is sent only by other NC servers
    // during synchronization session if some blob was written on this server
    // and the same data didn't make it to that server yet.
    { "SYNC_GET",
        {&CNCMessageHandler::x_DoCmd_SyncGet,
            "SYNC_GET",
            eSyncBlobCmd,
            eNCReadData, eProxyNone},
          // Server id of the server managing the synchronization.
        { { "srv_id",  eNSPT_Int,  eNSPA_Required },
          // Slot that synchronization is started on.
          { "slot",    eNSPT_Int,  eNSPA_Required },
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
          { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Time of creation of initial record in synchronization log about
          // this operation.
          { "log_time",eNSPT_Int,  eNSPA_Required },
          // Creation time of the blob (microseconds since epoch).
          { "cr_time", eNSPT_Int,  eNSPA_Required },
          // Server_id of the server where blob was created.
          { "cr_srv",  eNSPT_Int,  eNSPA_Required },
          // Id of the blob on the server where it was created.
          { "cr_id",   eNSPT_Int,  eNSPA_Required } } },
    // Get information necessary to prolong the blob's life. This command
    // is sent only by other NC servers during synchronization session if some
    // blob was prolonged on this server and the same prolongation didn't
    // happen on that server yet.
    { "SYNC_PROINFO",
        {&CNCMessageHandler::x_DoCmd_SyncProlongInfo,
            "SYNC_PROINFO",
            eSyncBlobCmd,
            eNCRead, eProxyNone},
          // Server id of the server managing the synchronization.
        { { "srv_id",  eNSPT_Int,  eNSPA_Required },
          // Slot that synchronization is started on.
          { "slot",    eNSPT_Int,  eNSPA_Required },
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
          { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required } } },
    // "Commit" the synchronization session. This command is sent only by other
    // NC servers at the end of synchronization session when all necessary
    // commands have been executed successfully.
    { "SYNC_COMMIT",
        {&CNCMessageHandler::x_DoCmd_SyncCommit,
            "SYNC_COMMIT",
            eRunsInStartedSync | fProhibitsSyncAbort, eNCNone, eProxyNone},
          // Server id of the server managing the synchronization.
        { { "srv_id",  eNSPT_Int,  eNSPA_Required },
          // Slot that synchronization is started on.
          { "slot",    eNSPT_Int,  eNSPA_Required },
          // Last synchronized record number (in sync log) of _that_ server.
          { "rec_my",  eNSPT_Int,  eNSPA_Required },
          // Last synchronized record number (in sync log) of _this_ server.
          { "rec_your",eNSPT_Int,  eNSPA_Required } } },
    // "Cancel" the synchronization session. This command is sent only by other
    // NC servers at the end of synchronization session when either this server
    // requested or that server decided that synchronization should be aborted
    // despite the successful execution of all commands. Reason for
    // cancellation could be some server going to shutdown, or requirement to
    // clean sync logs (and synchronization already executes for too long).
    // The cancellation doesn't cancel any commands already executed in this
    // synchronization session. It exists only to quickly mark this
    // synchronization as no longer executing so that NC could start
    // synchronization with some other server.
    { "SYNC_CANCEL",
        {&CNCMessageHandler::x_DoCmd_SyncCancel,
            "SYNC_CANCEL",
            eRunsInStartedSync | fProhibitsSyncAbort, eNCNone, eProxyNone},
          // Server id of the server managing the synchronization.
        { { "srv_id",  eNSPT_Int,  eNSPA_Required },
          // Slot that synchronization is started on.
          { "slot",    eNSPT_Int,  eNSPA_Required } } },
    // Get meta information about the blob. Command for "NetCache" clients.
    { "GETMETA",
        {&CNCMessageHandler::x_DoCmd_GetMeta,
            "GETMETA",
            eClientBlobRead | fDoNotCheckPassword,
            eNCRead,
            eProxyGetMeta},
          // Key of the blob
        { { "key",     eNSPT_NCID, eNSPA_Required },
          // Flag forcing local execution of the command (without forwarding
          // to other servers and without searching for blob on them).
          { "local",   eNSPT_Int,  eNSPA_Optional },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Get meta information about the blob. Command for "ICache" clients.
    { "GETMETA",
        {&CNCMessageHandler::x_DoCmd_GetMeta,
            "IC_GETMETA",
            eClientBlobRead | fDoNotCheckPassword,
            eNCRead,
            eProxyGetMeta},
          // Name of cache for blob.
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's version.
          { "version", eNSPT_Int,  eNSPA_Required },
          // Blob's subkey.
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Flag forcing local execution of the command (without forwarding
          // to other servers and without searching for blob on them).
          { "local",   eNSPT_Int,  eNSPA_Optional },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Prolong blob's life for a specific number of seconds from current time.
    // Command provides a minimum time this blob should be still available for.
    // If blob's expiration time was already later than that this command is
    // a no-op.
    { "PROLONG",
        {&CNCMessageHandler::x_DoCmd_Prolong,
            "PROLONG",
            eClientBlobRead,
            eNCRead,
            eProxyProlong},
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Period of time for the blob to be available.
          { "ttl",     eNSPT_Int,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Optional },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Prolong blob's life for a specific number of seconds from current time.
    // This command is sent only by other NC servers in response to client's
    // PROLONG command when the server where client have sent initial command
    // cannot execute it locally for any reason (slot is not processed by that
    // server, or initial database caching is not completed yet, or it was
    // determined that this server has latest version of the blob).
    { "PROXY_PROLONG",
        {&CNCMessageHandler::x_DoCmd_Prolong,
            "PROXY_PROLONG",
            eProxyBlobRead | fDoNotProxyToPeers,
            eNCRead,
            eProxyProlong},
          // Name of cache for blob (for NC-generated blob keys this will be
          // empty).
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey (for NC-generated blob keys this will be empty).
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          // Period of time for the blob to be available.
          { "ttl",     eNSPT_Int,  eNSPA_Required },
          // Quorum to use for this operation.
          { "qrum",    eNSPT_Int,  eNSPA_Required },
          // Flag whether local execution of this command should be forced
          // no matter what, i.e. if this flag is set in cases when normal GET2
          // command would have been proxied to other servers an error should
          // be returned.
          { "local",   eNSPT_Int,  eNSPA_Required },
          // Value of the flag "search_on_read" to use with this command, i.e.
          // whether this server should search the blob if it's not found
          // locally.
          { "srch",    eNSPT_Int,  eNSPA_Required },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  eNSPA_Required },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Required },
          // Password for blob access.
          { "pass",    eNSPT_Str,  eNSPA_Optional } } },
    /*{ "BLOBSLIST",
        {&CNCMessageHandler::x_DoCmd_GetBlobsList,
            "BLOBSLIST",
            fNeedsStorageCache + fNeedsAdminClient} },*/
    // Write blob contents. Deprecated command used now only by old clients.
    // This command is the same as PUT3 except it uses connection closing as
    // legitimate EOF marker for blob's data.
    { "PUT2",
        {&CNCMessageHandler::x_DoCmd_Put,
            "PUT2",
            eClientBlobWrite | fCanGenerateKey | fCursedPUT2Cmd,
            eNCCreate,
            eProxyWrite},
          // Time-to-live for the blob.
        { { "ttl",     eNSPT_Int,  eNSPA_Optional },
          // Key of the blob (if skipped or empty then new one will be created).
          { "key",     eNSPT_NCID, eNSPA_Optional } } },
    // Shutdown the server
    { "SHUTDOWN",
        {&CNCMessageHandler::x_DoCmd_Shutdown,
            "SHUTDOWN",
            fNeedsAdminClient, eNCNone, eProxyNone},
          // Client IP for application sending the command.
        { { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // drain: wait until all BLOBs are expired, then shutdown
          { "drain",    eNSPT_Int,  eNSPA_Optional },
          // reset: shutdown and leave database guard on disk (CXX-10401)
          { "reset",    eNSPT_Int,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Get server statistics.
    { "GETSTAT",
        {&CNCMessageHandler::x_DoCmd_GetStat, "GETSTAT"},
          // Flag showing whether current (value is 0) or previous (value is 1)
          // statistics period should be shown.
        { { "prev",    eNSPT_Int,  fNSPA_Optional, "0" },
          // Type of statistics period to show. See top of nc_stat.cpp for
          // list of all possible period types.
          { "type",    eNSPT_Str,  eNSPA_Optional, "life" },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Read full ini-file used by NetCache for configuration.
    { "GETCONF",
        {&CNCMessageHandler::x_DoCmd_GetConfig,      "GETCONF",
            fNoCmdFlags, eNCNone, eProxyNone},
        { 
          // Netcached.ini section name
          { "section", eNSPT_Str,  eNSPA_Optional },
          // when section name is "netcache", setup for this port
          { "port",    eNSPT_Str,  eNSPA_Optional },
          // when section name is "netcache", setup for this cache
          { "cache",   eNSPT_Str,  eNSPA_Optional },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Get state information,  added in v6.8.6
    { "INFO",
        {&CNCMessageHandler::x_DoCmd_GetConfig,      "INFO",
            fNoCmdFlags, eNCNone, eProxyNone},
        { 
          // Netcached.ini section name
          { "section", eNSPT_Str,  fNSPA_Required },
          // when section name is "netcache", setup for this port
          { "port",    eNSPT_Str,  eNSPA_Optional },
          // when section name is "netcache", setup for this cache
          { "cache",   eNSPT_Str,  eNSPA_Optional },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Acknowledge alert,  added in v6.8.6
    { "ACKALERT",
        {&CNCMessageHandler::x_DoCmd_AckAlert,  "ACKALERT",
            fNeedsAdminClient, eNCNone, eProxyNone},
        { 
          // Alert name
          { "alert", eNSPT_Str,  fNSPA_Required },
          // User name
          { "user", eNSPT_Str,  fNSPA_Required },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Re-read ini-file used by NetCache for configuration.
    // Find changes and reconfigure
    // Only few changes are supported
    { "RECONF",
        {&CNCMessageHandler::x_DoCmd_ReConfig,
            "RECONF",
            fNeedsAdminClient, eNCNone, eProxyNone},
        { 
          // Netcached.ini section name
          { "section", eNSPT_Str,  eNSPA_Required },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
 
    { "PURGE",
        {&CNCMessageHandler::x_DoCmd_Purge,
            "PURGE",
            fNoCmdFlags, eNCNone, eProxyNone},
        { 
            // Cache name.
            { "cache",     eNSPT_Str, eNSPA_Required },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    { "COPY_PURGE",
        {&CNCMessageHandler::x_DoCmd_CopyPurge,
            "COPY_PURGE", fNoCmdFlags, eNCNone, eProxyNone },
          // Cache name.
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // forget blobs created earlier than cr_time
          { "cr_time", eNSPT_Int,  eNSPA_Required } } },
    
    // Added in 6.11.7, CXX-8948
    { "PURGE2",
        {&CNCMessageHandler::x_DoCmd_Purge,
            "PURGE",
            fNoCmdFlags, eNCNone, eProxyNone},
          // Cache name.
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    // Added in 6.11.7, CXX-8948
    { "COPY_PURGE2",
        {&CNCMessageHandler::x_DoCmd_CopyPurge,
            "COPY_PURGE", fNoCmdFlags, eNCNone, eProxyNone },
          // Cache name.
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // forget blobs created earlier than cr_time
          { "cr_time", eNSPT_Int,  eNSPA_Required } } },
 
    // Get list of blobs by mask: "cache,key,*"
    // Added in 6.9.0, CXX-6246
    { "BLIST",
        {&CNCMessageHandler::x_DoCmd_GetBList,
            "BLIST",
            fComesFromClient |
                fNeedsStorageCache | fNeedsBlobList | fDoNotCheckPassword,
            eNCNone, eProxyGetBList},
          // Name of cache for blob.
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey.
          { "subkey",  eNSPT_Str,  eNSPA_Optional },
          { "local",   eNSPT_Int,  eNSPA_Optional },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
// added in v6.11.0 (CXX-8737)
    { "BLIST2",
        {&CNCMessageHandler::x_DoCmd_GetBList,
            "BLIST2",
            fComesFromClient |
                fNeedsStorageCache | fNeedsBlobList | fDoNotProxyToPeers | fDoNotCheckPassword,
            eNCNone, eProxyGetBList2},
          // Name of cache for blob.
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Optional },
          // Blob's subkey.
          { "subkey",  eNSPT_Str,  eNSPA_Optional },
          { "local",   eNSPT_Int,  eNSPA_Optional },
 
          // Created more than N seconds ago
          { "fcr_ago_ge", eNSPT_Int,  eNSPA_Optional },
          // Created less than N seconds ago
          { "fcr_ago_lt", eNSPT_Int,  eNSPA_Optional },
          // Created more than N seconds since epoch
          { "fcr_epoch_ge", eNSPT_Int,  eNSPA_Optional },
          // Created less than N seconds since epoch
          { "fcr_epoch_lt", eNSPT_Int,  eNSPA_Optional },
          // Blob will expire in more than N seconds from now
          { "fexp_now_ge", eNSPT_Int,  eNSPA_Optional },
          // Blob will expire in less than N seconds from now
          { "fexp_now_lt", eNSPT_Int,  eNSPA_Optional },
          // Blob will expire in more than N seconds since epoch
          { "fexp_epoch_ge", eNSPT_Int,  eNSPA_Optional },
          // Blob will expire in less than N seconds since epoch
          { "fexp_epoch_lt", eNSPT_Int,  eNSPA_Optional },
          // Version will expire in more than N seconds from now
          { "fvexp_now_ge", eNSPT_Int,  eNSPA_Optional },
          // Version will expire in less than N seconds from now
          { "fvexp_now_lt", eNSPT_Int,  eNSPA_Optional },
          // Version will expire in more than N seconds since epoch
          { "fvexp_epoch_ge", eNSPT_Int,  eNSPA_Optional },
          // Version will expire in less than N seconds since epoch
          { "fvexp_epoch_lt", eNSPT_Int,  eNSPA_Optional },
          // Server_id of the server where blob was created.
          { "fcr_srv",  eNSPT_Int,  eNSPA_Optional },
          // blob bigger than this size
          { "fsize_ge",  eNSPT_Int,  eNSPA_Optional },
          // blob smaller than this size
          { "fsize_lt",  eNSPT_Int,  eNSPA_Optional },
          // Client IP for application sending the command.
          { "ip",      eNSPT_Str,  fNSPA_Optional },
          // Session ID for application sending the command.
          { "sid",     eNSPT_Str,  eNSPA_Optional },
          // request Hit ID
          { "ncbi_phid", eNSPT_Str,  eNSPA_Optional }
        } },
    { "PROXY_BLIST",
        {&CNCMessageHandler::x_DoCmd_GetBList,
            "PROXY_BLIST",
            fNeedsStorageCache | fNeedsBlobList | fDoNotProxyToPeers | fDoNotCheckPassword,
            eNCNone, eProxyNone},
          // Name of cache for blob.
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey.
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          { "local",   eNSPT_Int,  eNSPA_Required }
        } },
// added in v6.11.0 (CXX-8737)
    { "PROXY_BLIST2",
        {&CNCMessageHandler::x_DoCmd_GetBList,
            "PROXY_BLIST2",
            fNeedsStorageCache | fNeedsBlobList | fDoNotProxyToPeers | fDoNotCheckPassword,
            eNCNone, eProxyNone},
          // Name of cache for blob.
        { { "cache",   eNSPT_Str,  eNSPA_Required },
          // Blob's key.
          { "key",     eNSPT_Str,  eNSPA_Required },
          // Blob's subkey.
          { "subkey",  eNSPT_Str,  eNSPA_Required },
          { "local",   eNSPT_Int,  eNSPA_Required },
          // Created more than N seconds ago
          { "fcr_ago_ge", eNSPT_Int,  eNSPA_Optional },
          // Created less than N seconds ago
          { "fcr_ago_lt", eNSPT_Int,  eNSPA_Optional },
          // Created more than N seconds since epoch
          { "fcr_epoch_ge", eNSPT_Int,  eNSPA_Optional },
          // Created less than N seconds since epoch
          { "fcr_epoch_lt", eNSPT_Int,  eNSPA_Optional },
          // Will expire in more than N seconds from now
          { "fexp_now_ge", eNSPT_Int,  eNSPA_Optional },
          // Will expire in less than N seconds from now
          { "fexp_now_lt", eNSPT_Int,  eNSPA_Optional },
          // Will expire in more than N seconds since epoch
          { "fexp_epoch_ge", eNSPT_Int,  eNSPA_Optional },
          // Will expire in less than N seconds since epoch
          { "fexp_epoch_lt", eNSPT_Int,  eNSPA_Optional },
          // Will expire in more than N seconds from now
          { "fvexp_now_ge", eNSPT_Int,  eNSPA_Optional },
          // Will expire in less than N seconds from now
          { "fvexp_now_lt", eNSPT_Int,  eNSPA_Optional },
          // Will expire in more than N seconds since epoch
          { "fvexp_epoch_ge", eNSPT_Int,  eNSPA_Optional },
          // Will expire in less than N seconds since epoch
          { "fvexp_epoch_lt", eNSPT_Int,  eNSPA_Optional },
          // Server_id of the server where blob was created.
          { "fcr_srv",  eNSPT_Int,  eNSPA_Optional },
          // blob bigger than this size
          { "fsize_ge",  eNSPT_Int,  eNSPA_Optional },
          // blob smaller than this size
          { "fsize_lt",  eNSPT_Int,  eNSPA_Optional }
        } },
 
// HTTP commands
    { "DELETE",
        {&CNCMessageHandler::x_DoCmd_Remove,
            "DELETE",
            fNeedsBlobAccess | fNoBlobAccessStats | fIsHttp,
            eNCCreate,
            eProxyRemove}
    },
    { "GET",
        {&CNCMessageHandler::x_DoCmd_Get,
            "GET",
            eClientBlobRead | fIsHttp,
            eNCReadData,
            eProxyRead}
    },
    { "HEAD",
        {&CNCMessageHandler::x_DoCmd_GetMeta,
            "HEAD",
            eClientBlobRead | fDoNotCheckPassword | fIsHttp,
            eNCRead,
            eProxyGetMeta}
    },
    { "POST",
        {&CNCMessageHandler::x_DoCmd_Put,
            "POST",
            eClientBlobWrite | fCanGenerateKey | fReadExactBlobSize | fSkipBlobEOF | fIsHttp,
            eNCCreate,
            eProxyWrite}
    },
    { "PUT",
        {&CNCMessageHandler::x_DoCmd_Put,
            "PUT",
            eClientBlobWrite | fIsHttp | fReadExactBlobSize | fSkipBlobEOF | fIsHttp,
            eNCCreate,
            eProxyWrite}
    },
 
 
    // All commands below are not implemented and mostly old ones not needed
    // anymore. One exception is RECONF - it would be nice to have it but it
    // needs some thinking on how to implement it.
 
    { "REINIT",   {&CNCMessageHandler::x_DoCmd_NotImplemented, "REINIT"} },
    { "REINIT",   {&CNCMessageHandler::x_DoCmd_NotImplemented, "IC_REINIT"},
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix } } },
//    { "RECONF",   {&CNCMessageHandler::x_DoCmd_NotImplemented, "RECONF"} },
    { "LOG",      {&CNCMessageHandler::x_DoCmd_NotImplemented, "LOG"} },
    { "STAT",     {&CNCMessageHandler::x_DoCmd_NotImplemented, "STAT"} },
    { "MONI",     {&CNCMessageHandler::x_DoCmd_NotImplemented, "MONITOR"} },
    { "DROPSTAT", {&CNCMessageHandler::x_DoCmd_NotImplemented, "DROPSTAT"} },
    { "GBOW",     {&CNCMessageHandler::x_DoCmd_NotImplemented, "GBOW"} },
    { "ISLK",     {&CNCMessageHandler::x_DoCmd_NotImplemented, "ISLK"} },
    { "SMR",      {&CNCMessageHandler::x_DoCmd_NotImplemented, "SMR"} },
    { "SMU",      {&CNCMessageHandler::x_DoCmd_NotImplemented, "SMU"} },
    { "OK",       {&CNCMessageHandler::x_DoCmd_NotImplemented, "OK"} },
// will be used by HTTP
//    { "GET",      {&CNCMessageHandler::x_DoCmd_NotImplemented, "GET"} },
//    { "PUT",      {&CNCMessageHandler::x_DoCmd_NotImplemented, "PUT"} },
    { "REMOVE",   {&CNCMessageHandler::x_DoCmd_NotImplemented, "REMOVE"} },
    { "STSP",     {&CNCMessageHandler::x_DoCmd_NotImplemented, "IC_STSP"},
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix } } },
    { "GTSP",     {&CNCMessageHandler::x_DoCmd_NotImplemented, "IC_GTSP"},
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix } } },
    { "SVRP",     {&CNCMessageHandler::x_DoCmd_NotImplemented, "IC_SVRP"},
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix } } },
    { "GVRP",     {&CNCMessageHandler::x_DoCmd_NotImplemented, "IC_GVRP"},
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix } } },
    { "PRG1",     {&CNCMessageHandler::x_DoCmd_NotImplemented, "IC_PRG1"},
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix } } },
    { "REMK",     {&CNCMessageHandler::x_DoCmd_NotImplemented, "IC_REMK"},
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix } } },
    { "GBLW",     {&CNCMessageHandler::x_DoCmd_NotImplemented, "IC_GBLW"},
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix } } },
    { "ISOP",     {&CNCMessageHandler::x_DoCmd_NotImplemented, "IC_ISOP"},
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix } } },
    { "GACT",     {&CNCMessageHandler::x_DoCmd_NotImplemented, "IC_GACT"},
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix } } },
    { "GTOU",     {&CNCMessageHandler::x_DoCmd_NotImplemented, "IC_GTOU"},
        { { "cache",   eNSPT_Id,   eNSPA_ICPrefix } } },
    { NULL }
};
 
// List of arguments that can be in client authentication line.
static SNSProtoArgument s_AuthArgs[] = {
    { "client", eNSPT_Str, eNSPA_Optional, "Unknown client" },
    { "params", eNSPT_Str, eNSPA_Ellipsis },
    { NULL }
};
 
 
 
 
/////////////////////////////////////////////////////////////////////////////
// CNCMessageHandler implementation
 
inline void
CNCMessageHandler::x_ResetFlags(void)
{
    TNCCmdFlags keep = m_Flags & fNoReplyOnFinish;
    m_Flags = fNoCmdFlags | keep;
}
 
inline void
CNCMessageHandler::x_SetFlag(ENCCmdFlags flag)
{
    m_Flags |= flag;
}
 
inline void
CNCMessageHandler::x_UnsetFlag(ENCCmdFlags flag)
{
    m_Flags &= ~flag;
}
 
inline bool
CNCMessageHandler::x_IsFlagSet(ENCCmdFlags flag)
{
    return (m_Flags & flag) != 0;
}
 
inline bool
CNCMessageHandler::x_IsUserFlagSet(ENCUserFlags flag)
{
    return (m_UserFlags & flag) != 0;
}
 
inline bool
CNCMessageHandler::x_IsCmdSucceeded(int cmd_status)
{
    return cmd_status < 300;
}
 
//static Uint8 s_CntHndls = 0;
 
CNCMessageHandler::CNCMessageHandler(void)
    : m_Flags(0),
      m_Parser(s_CommandMap),
      m_CmdProcessor(NULL),
      m_BlobAccess(NULL),
      m_write_event(NULL),
      m_ChunkLen(0),
      m_SrvsIndex(0),
      m_ActiveHub(NULL)
{
    LOG_CURRENT_FUNCTION
#if __NC_TASKS_MONITOR
    m_TaskName = "CNCMessageHandler";
#endif
    m_CopyBlobInfo = new SNCBlobVerData(nullptr);
    m_LatestBlobSum = new SNCBlobSummary();
    m_BlobFilter = new SNCBlobFilter;
    m_CmdLog.reserve(32);
 
    SetState(&CNCMessageHandler::x_SocketOpened);
 
    //Uint8 cnt = AtomicAdd(s_CntHndls, 1);
    //INFO("CNCMessageHandler, cnt=" << cnt);
    m_HttpMode = eNoHttp;
}
 
CNCMessageHandler::~CNCMessageHandler(void)
{
    LOG_CURRENT_FUNCTION
    delete m_LatestBlobSum;
    delete m_CopyBlobInfo;
 
    //Uint8 cnt = AtomicSub(s_CntHndls, 1);
    //INFO("~CNCMessageHandler, cnt=" << cnt);
}
 
CNCMessageHandler::State
CNCMessageHandler::x_SocketOpened(void)
{
    LOG_CURRENT_FUNCTION
    m_PrevCache.clear();
    m_ClientParams.clear();
    m_CntCmds = 0;
 
    string host;
    Uint2 port = 0;
    GetPeerAddress(host, port);
    m_ClientParams["peer"]  = host;
    m_ClientParams["pport"] = NStr::UIntToString(port);
    m_LocalPort             = GetLocalPort();
    m_ClientParams["port"]  = NStr::UIntToString(m_LocalPort);
 
    return &CNCMessageHandler::x_ReadAuthMessage;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_CloseCmdAndConn(void)
{
    LOG_CURRENT_FUNCTION
    if (GetDiagCtx()->GetRequestStatus() == eStatus_OK) {
        if (HasError()  ||  !CanHaveMoreRead())
            GetDiagCtx()->SetRequestStatus(eStatus_PrematureClose);
        else if (CTaskServer::IsInShutdown())
            GetDiagCtx()->SetRequestStatus(eStatus_ShuttingDown);
        else
            GetDiagCtx()->SetRequestStatus(eStatus_CmdTimeout);
    }
    int status = GetDiagCtx()->GetRequestStatus();
    if (x_IsFlagSet(eBlobPut) && m_NCBlobKey.IsICacheKey()) {
        x_JournalBlobPutResult(status, m_NCBlobKey.PackedKey(), m_BlobSlot);
    }
    x_CleanCmdResources();
    GetDiagCtx()->SetRequestStatus(status);
    return &CNCMessageHandler::x_SaveStatsAndClose;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_SaveStatsAndClose(void)
{
    LOG_CURRENT_FUNCTION
    //CNCStat::AddClosedConnection(conn_span, GetDiagCtx()->GetRequestStatus(), m_CntCmds);
    CNCStat::ConnClosing(m_CntCmds);
    return &CNCMessageHandler::x_PrintCmdsCntAndClose;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_PrintCmdsCntAndClose(void)
{
    LOG_CURRENT_FUNCTION
    CSrvDiagMsg().PrintExtra().PrintParam("cmds_cnt", m_CntCmds);
    CloseSocket();
    Terminate();
    return NULL;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_WriteInitWriteResponse(void)
{
    LOG_CURRENT_FUNCTION
check_again:
    ENCClientHubStatus status = m_ActiveHub->GetStatus();
    if (status == eNCHubError || status == eNCHubSuccess) {
        m_LastPeerError = m_ActiveHub->GetErrMsg();
        SRV_LOG(Warning, "Error executing command on peer "
            << m_ActiveHub->GetFullPeerName() << ", peer says: "
            << m_LastPeerError);
        m_ActiveHub->Release();
        m_ActiveHub = NULL;
        return &CNCMessageHandler::x_ProxyToNextPeer;
    }
    else if (status != eNCHubCmdInProgress) {
        SRV_FATAL("Unexpected client status: " << status);
    }
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_CloseCmdAndConn;
 
    CNCActiveHandler* active = m_ActiveHub->GetHandler();
    if (!active->GotClientResponse())
        return NULL;
    // Intentionally re-reading the status because it could change since our
    // previous read at the beginning of the function.
    if (m_ActiveHub->GetStatus() != eNCHubCmdInProgress)
        goto check_again;
 
    WriteText(active->GetCmdResponse()).WriteText("\n");
    Flush();
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_CloseCmdAndConn;
 
    return &CNCMessageHandler::x_ReadBlobSignature;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_ReadAuthMessage(void)
{
    LOG_CURRENT_FUNCTION
    if (NeedToClose()  ||  CTaskServer::IsInSoftShutdown()) {
        if (CTaskServer::IsInShutdown())
            GetDiagCtx()->SetRequestStatus(eStatus_ShuttingDown);
        else
            GetDiagCtx()->SetRequestStatus(eStatus_CmdTimeout);
        return &CNCMessageHandler::x_SaveStatsAndClose;
    }
 
    CTempString auth_line;
    if (!ReadLine(&auth_line)) {
        if (!HasError()  &&  CanHaveMoreRead())
            return NULL;
        if (IsReadDataAvailable()) {
            GetDiagCtx()->SetRequestStatus(eStatus_PrematureClose);
            return &CNCMessageHandler::x_SaveStatsAndClose;
        }
        else {
            GetDiagCtx()->SetRequestStatus(eStatus_CmdTimeout);
            return &CNCMessageHandler::x_PrintCmdsCntAndClose;
        }
    }
 
    m_HttpMode = eNoHttp;
    size_t auth_size = auth_line.size();
    if (auth_size > 8 && auth_line[auth_size-8] == 'H') {
        if (NStr::strncmp(auth_line.data() + auth_size - 8, "HTTP/1.", 7) == 0) {
            if (auth_line[auth_size-1] == '0') {
                m_HttpMode = eHttp10;
            } else if (auth_line[auth_size-1] == '1') {
                m_HttpMode = eHttp11;
            }
        }
    }
 
    if (!x_IsHttpMode()) {
        TNSProtoParams params;
        try {
            m_Parser.ParseArguments(auth_line, s_AuthArgs, &params);
        }
        catch (CNSProtoParserException& ex) {
            SRV_LOG(Error, "Error authenticating client: '"
                             << auth_line << "': " << ex);
            params["client"] = auth_line;
            return &CNCMessageHandler::x_SaveStatsAndClose;
        }
        ITERATE(TNSProtoParams, it, params) {
            m_ClientParams[it->first] = it->second;
        }
        CSrvDiagMsg diag_msg;
        diag_msg.PrintExtra();
        ITERATE(TNSProtoParams, it, params) {
            diag_msg.PrintParam(it->first, it->second);
        }
        diag_msg.Flush();
    } else {
        m_PosponedCmd = auth_line;
//        m_ClientParams["cache"] = "http";
//        m_ClientParams["cache"] = "";
    }
 
    m_BaseAppSetup = m_AppSetup = CNCServer::GetAppSetup(m_ClientParams);
    if (m_AppSetup->disable) {
        SRV_LOG(Warning, "Disabled client is being disconnected ('"
                         << auth_line << "').");
        GetDiagCtx()->SetRequestStatus(eStatus_Disabled);
        return &CNCMessageHandler::x_SaveStatsAndClose;
    }
    else {
        return &CNCMessageHandler::x_ReadCommand;
    }
}
 
void
CNCMessageHandler::x_AssignCmdParams(void)
{
    LOG_CURRENT_FUNCTION
    CTempString blob_key, blob_subkey;
    m_UserFlags = fNoUserFlags;
    m_BlobVersion = 0;
    m_NCBlobKey.Clear();
    m_BlobPass.clear();
    m_RawBlobPass.clear();
    m_KeyVersion = 1;
    m_BlobTTL = 0;
    m_StartPos = 0;
    m_Size = Uint8(-1);
    m_Slot = 0;
    m_OrigRecNo = 0;
    m_OrigSrvId = 0;
    m_OrigTime = 0;
    m_Quorum = 1;
    m_CmdVersion = 0;
    m_ForceLocal = false;
    m_AgeMax = m_AgeCur = 0;
    m_SlotsDone.clear();
    m_CmdParams.clear();
    bool quorum_was_set = false;
    bool search_was_set = false;
 
    new (m_LatestBlobSum) SNCBlobSummary();
    delete m_CopyBlobInfo;
    m_CopyBlobInfo = new SNCBlobVerData(nullptr);
    new (m_BlobFilter) SNCBlobFilter();
 
    CTempString cache_name;
 
    if (!x_IsHttpMode()) {
        m_HttpMode = eNoHttp;
        ERASE_ITERATE(TNSProtoParams, it, m_ParsedCmd.params) {
            const CTempString& key = it->first;
            CTempString& val = it->second;
 
            switch (key[0]) {
            case 'a':
                if (key == "age") {
                    m_AgeMax = NStr::StringToUInt8(val);
                }
                break;
            case 'c':
                switch (key[1]) {
                case 'a':
                    if (key == "cache") {
                        cache_name = val;
                    }
                    break;
                case 'm':
                    if (key == "cmd_ver") {
                        m_CmdVersion = NStr::StringToUInt(val);
                    }
                    break;
                case 'o':
                    if (key == "confirm") {
                        if (val == "1")
                            x_UnsetFlag(fNoReplyOnFinish);
                        else
                            x_SetFlag(fNoReplyOnFinish);
                    }
                    break;
                case 'r':
                    if (key == "cr_time") {
                        m_CopyBlobInfo->create_time = NStr::StringToUInt8(val);
                    }
                    else if (key == "cr_id") {
                        m_CopyBlobInfo->create_id = NStr::StringToUInt(val);
                    }
                    else if (key == "cr_srv") {
                        m_CopyBlobInfo->create_server = NStr::StringToUInt8(val);
                    }
                    break;
                }
                break;
            case 'd':
                if (key == "dead") {
                    m_CopyBlobInfo->dead_time = NStr::StringToInt(val);
                }
                break;
            case 'e':
                if (key == "exp") {
                    m_CopyBlobInfo->expire = NStr::StringToInt(val);
                }
                break;
            case 'f':
                if (key == "flags") {
                    m_UserFlags = NStr::StringToUInt(val);
                } else if (key == "fcr_ago_ge") {
                    m_BlobFilter->cr_ago_ge = NStr::StringToUInt8(val);
                } else if (key == "fcr_ago_lt") {
                    m_BlobFilter->cr_ago_lt = NStr::StringToUInt8(val);
                } else if (key == "fcr_epoch_ge") {
                    m_BlobFilter->cr_epoch_ge = NStr::StringToUInt8(val);
                } else if (key == "fcr_epoch_lt") {
                    m_BlobFilter->cr_epoch_lt = NStr::StringToUInt8(val);
                } else if (key == "fexp_now_ge") {
                    m_BlobFilter->exp_now_ge = NStr::StringToUInt8(val);
                } else if (key == "fexp_now_lt") {
                    m_BlobFilter->exp_now_lt = NStr::StringToUInt8(val);
                } else if (key == "fexp_epoch_ge") {
                    m_BlobFilter->exp_epoch_ge = NStr::StringToUInt8(val);
                } else if (key == "fexp_epoch_lt") {
                    m_BlobFilter->exp_epoch_lt = NStr::StringToUInt8(val);
                } else if (key == "fvexp_now_ge") {
                    m_BlobFilter->vexp_now_ge = NStr::StringToUInt8(val);
                } else if (key == "fvexp_now_lt") {
                    m_BlobFilter->vexp_now_lt = NStr::StringToUInt8(val);
                } else if (key == "fvexp_epoch_ge") {
                    m_BlobFilter->vexp_epoch_ge = NStr::StringToUInt8(val);
                } else if (key == "fvexp_epoch_lt") {
                    m_BlobFilter->vexp_epoch_lt = NStr::StringToUInt8(val);
                } else if (key == "fcr_srv") {
                    m_BlobFilter->cr_srv = NStr::StringToUInt8(val);
                } else if (key == "fsize_ge") {
                    m_BlobFilter->size_ge = NStr::StringToUInt8(val);
                } else if (key == "fsize_lt") {
                    m_BlobFilter->size_lt = NStr::StringToUInt8(val);
                }
                break;
            case 'h':
                if (key == "http") {
                    m_HttpMode = (EHttpMode)NStr::StringToInt(val);
                }
                break;
            case 'i':
                if (key == "ip") {
                    if (!val.empty())
                        GetDiagCtx()->SetClientIP(val);
                    // Erase parameter to not print it in request-start, it will be
                    // printed as a part of standard log header.
                    m_ParsedCmd.params.erase(it);
                }
                break;
            case 'k':
                if (key == "key") {
                    blob_key = val;
                }
                break;
            case 'l':
                if (key == "log_rec") {
                    m_OrigRecNo = NStr::StringToUInt8(val);
                }
                else if (key == "log_srv") {
                    m_OrigSrvId = NStr::StringToUInt8(val);
                }
                else if (key == "log_time") {
                    m_OrigTime = NStr::StringToUInt8(val);
                }
                else if (key == "local") {
                    m_ForceLocal = val == "1";
                }
                break;
            case 'm':
                if (key == "md5_pass") {
                    m_BlobPass = val;
                }
                break;
            case 'n':
                if (key == "ncbi_phid") {
                    if (!val.empty()) {
                        GetDiagCtx()->SetHitID(val);
                    }
                }
                break;
            case 'p':
                if (key == "pass") {
                    m_RawBlobPass = val;
                    CMD5 md5;
                    md5.Update(val.data(), val.size());
                    unsigned char digest[16];
                    md5.Finalize(digest);
                    m_BlobPass.assign((char*)digest, 16);
                    // Erase parameter to not expose passwords via logs.
                    m_ParsedCmd.params.erase(it);
                }
                else if (key == "prev") {
                    m_StatPrev = val == "1";
                }
                break;
            case 'q':
                if (key == "qrum") {
                    m_Quorum = NStr::StringToUInt(val);
                    quorum_was_set = true;
                }
                break;
            case 'r':
                if (key == "rec_my") {
                    m_RemoteRecNo = NStr::StringToUInt8(val);
                }
                else if (key == "rec_your") {
                    m_LocalRecNo = NStr::StringToUInt8(val);
                }
                break;
            case 's':
                switch (key[1]) {
                case 'i':
                    if (key == "sid") {
                        if (!val.empty())
                            GetDiagCtx()->SetSessionID(NStr::URLDecode(val));
                        // Erase parameter to not print it in request-start,
                        // it will be printed as a part of standard log header.
                        m_ParsedCmd.params.erase(it);
                    }
                    else if (key == "size") {
                        m_Size = Uint8(NStr::StringToInt8(val));
                    }
                    break;
                case 'l':
                    if (key == "slot") {
                        m_Slot = Uint2(NStr::StringToUInt(val));
                    }
                    break;
                case 'r':
                    if (key == "srv_id") {
                        m_SrvId = NStr::StringToUInt8(val);
                    }
                    else if (key == "srch") {
                        m_SearchOnRead = val != "0";
                        search_was_set = true;
                    }
                    break;
                case 't':
                    if (key == "start") {
                        m_StartPos = NStr::StringToUInt8(val);
                    }
                    break;
                case 'u':
                    if (key == "subkey") {
                        blob_subkey = val;
                    }
                    break;
                }
                break;
            case 't':
                if (key == "ttl") {
                    m_BlobTTL = NStr::StringToUInt(val);
                }
                else if (key == "type") {
                    m_StatType = val;
                }
                break;
            case 'v':
                if (key == "version") {
                    m_BlobVersion = NStr::StringToInt(val);
                }
                else if (key == "ver_ttl") {
                    m_CopyBlobInfo->ver_ttl = NStr::StringToUInt(val);
                }
                else if (key == "ver_dead") {
                    m_CopyBlobInfo->ver_expire = NStr::StringToInt(val);
                }
                break;
            default:
                break;
            }
        }
    } else {
        m_ClientParams["client"] = "";
        cache_name = m_ClientParams["cache"];
        if (m_ClientParams.find("key") != m_ClientParams.end()) {
            blob_key = m_ClientParams["key"];
        }
        // parse HTTP header
        CTempString cmd_line;
        map<string,string> headers;
        while (ReadLine(&cmd_line)) {
            if (cmd_line.empty()) {
                break;
            }
            const string content_length("Content-Length:");
            const string     user_agent("User-Agent:");
            const string  content_range("Range:");
            size_t max_pos = cmd_line.size();
 
            if (NStr::StartsWith(cmd_line, content_length)) {
                size_t pos = content_length.size();
                m_Size = NStr::StringToUInt8(
                    CTempString(cmd_line.data() + pos, max_pos - pos),
                    NStr::fAllowLeadingSpaces | NStr::fAllowTrailingSpaces);
            }
            else if (NStr::StartsWith(cmd_line, content_range)) {
                size_t pos = content_range.size();
                const char* begin = cmd_line.data() + pos;
                while (!isdigit(*begin) && pos < max_pos) {
                    ++pos; ++begin;
                }
                m_StartPos = 0;
                while (isdigit(*begin) && pos < max_pos) {
                    m_StartPos = m_StartPos * 10 + (*begin - '0');
                    ++pos; ++begin;
                }
                while (!isdigit(*begin) && pos < max_pos) {
                    ++pos; ++begin;
                }
                m_Size = 0;
                while (isdigit(*begin) && pos < max_pos) {
                    m_Size = m_Size * 10 + (*begin - '0');
                    ++pos; ++begin;
                }
                // byte pos are inclusive
                m_Size = (m_Size >= m_StartPos) ? (m_Size - m_StartPos + 1) : 0;
            }
            else if (NStr::StartsWith(cmd_line, user_agent)) {
                list<CTempString>   arr;
                size_t pos = user_agent.size();
                NStr::Split( CTempString(cmd_line.data()+pos, max_pos-pos), " ", arr, NStr::fSplit_Tokenize);
                if (!arr.empty()) {
                    m_ClientParams["client"] = NStr::URLEncode( arr.front());
                }
            }
            else {
                string key, value;
                NStr::SplitInTwo(cmd_line, ":", key, value);
                key = NStr::TruncateSpaces(key);
                key = NStr::ToUpper(key);
                value = NStr::TruncateSpaces(value);
                headers[key] = value;
            }
        }
        list<CTempString>   arr;
        if (headers.find("NCBI-SID") != headers.end()) {
            NStr::Split( headers.at("NCBI-SID"), ", \t", arr, NStr::fSplit_Tokenize);
            if (!arr.empty()) {
                m_CmdParams["sid"] = arr.back();
            }
        }
        arr.clear();
        if (headers.find("NCBI-PHID") != headers.end()) {
            NStr::Split( headers.at("NCBI-PHID"), ", \t", arr, NStr::fSplit_Tokenize);
            if (!arr.empty()) {
                m_CmdParams["ncbi_phid"] = arr.back();
            }
        }
        arr.clear();
        string client_ip = g_GetClientIP( headers);
        if (!client_ip.empty()) {
            NStr::Split( client_ip, ", \t", arr, NStr::fSplit_Tokenize);
        }
        if (!arr.empty()) {
            m_CmdParams["ip"] = arr.front();
        } else {
            m_CmdParams["ip"] = m_ClientParams["peer"];
        }
    }
 
    m_NCBlobKey.Assign(cache_name, blob_key, blob_subkey);
    if (cache_name.empty()) {
        m_AppSetup = m_BaseAppSetup;
        m_ClientParams.erase("cache");
    }
    else if (cache_name == m_PrevCache) {
        m_AppSetup = m_PrevAppSetup;
    }
    else {
        m_PrevCache = cache_name;
        m_ClientParams["cache"] = cache_name;
        m_AppSetup = m_PrevAppSetup = CNCServer::GetAppSetup(m_ClientParams);
    }
    if (!quorum_was_set)
        m_Quorum = m_AppSetup->quorum;
    if (m_ForceLocal)
        m_Quorum = 1;
    if (!search_was_set)
        m_SearchOnRead = m_AppSetup->srch_on_read;
}
 
void
CNCMessageHandler::x_PrintRequestStart(CSrvDiagMsg& diag_msg)
{
    LOG_CURRENT_FUNCTION
    diag_msg.StartRequest();
    diag_msg.PrintParam("_type", "cmd");
    diag_msg.PrintParam("cmd", m_ParsedCmd.command->cmd);
    diag_msg.PrintParam("client", m_ClientParams["client"]);
    diag_msg.PrintParam("conn", m_ConnReqId);
    diag_msg.PrintParam("phost", m_ClientParams["peer"]);
    ITERATE(TNSProtoParams, it, m_ParsedCmd.params) {
        diag_msg.PrintParam(it->first, it->second);
    }
    ITERATE(TStringMap, it, m_CmdParams) {
        diag_msg.PrintParam(it->first, it->second);
    }
    if (!m_BlobPass.empty()) {
        diag_msg.PrintParam("pass", CNCBlobStorage::PrintablePassword(m_BlobPass));
    }
    diag_msg.PrintParam("ncbi_role", CNCServer::GetHostRole());
    diag_msg.PrintParam("ncbi_location", CNCServer::GetHostLocation());
}
 
CNCMessageHandler::State
CNCMessageHandler::x_StartCommand(void)
{
    LOG_CURRENT_FUNCTION
    m_CmdPrevTime = m_CmdStartTime = CSrvTime::Current();
    m_CmdLog.clear();
    CNCStat::CmdStarted(m_ParsedCmd.command->cmd);
    CSrvDiagMsg diag_msg;
    x_PrintRequestStart(diag_msg);
 
    if (NeedToClose()) {
        diag_msg.Flush();
        x_ResetFlags();
        return &CNCMessageHandler::x_CloseCmdAndConn;
    }
    if (HasError()  ||  !CanHaveMoreRead()) {
        diag_msg.Flush();
        GetDiagCtx()->SetRequestStatus(eStatus_PrematureClose);
        x_ResetFlags();
        return &CNCMessageHandler::x_CloseCmdAndConn;
    }
 
    if (x_IsFlagSet(fNeedsAdminClient)
        &&  m_ClientParams["client"] != CNCServer::GetAdminClient()
        &&  m_ClientParams["client"] != kNCPeerClientName)
    {
        string msg;
        msg += "command: " + string(m_ParsedCmd.command->cmd);
        msg += ", peer: " + m_ClientParams["peer"];
        msg += ", client: " + m_ClientParams["client"];
        CNCAlerts::Register(CNCAlerts::eAccessDenied, msg);
        diag_msg.Flush();
        x_ReportError(eStatus_NeedAdmin);
        if (!x_IsHttpMode()) {
            x_ResetFlags();
        }
        return &CNCMessageHandler::x_FinishCommand;
    }
 
    if (!CNCServer::IsCachingComplete()
        &&  (x_IsFlagSet(fNeedsStorageCache)  ||  m_ForceLocal))
    {
        diag_msg.Flush();
        x_ReportError(eStatus_JustStarted);
        if (!x_IsHttpMode()) {
            x_ResetFlags();
        }
        return &CNCMessageHandler::x_FinishCommand;
    }
 
    if (m_AppSetup->disable) {
        diag_msg.Flush();
        // We'll be here only if generally work for the client is enabled but
        // for current particular cache it is disabled.
        x_ReportError(eStatus_Disabled);
        if (!x_IsHttpMode()) {
            x_ResetFlags();
        }
        return &CNCMessageHandler::x_FinishCommand;
    }
 
    if (x_IsFlagSet(fRunsInStartedSync)) {
        ESyncInitiateResult start_res = CNCPeriodicSync::CanStartSyncCommand(
                                        m_SrvId, m_Slot,
                                        !x_IsFlagSet(fProhibitsSyncAbort),
                                        m_SyncId);
        if (start_res == eNetworkError) {
            diag_msg.Flush();
            x_ReportError(eStatus_CmdAborted);
            if (!x_IsHttpMode()) {
                x_ResetFlags();
            }
            return &CNCMessageHandler::x_FinishCommand;
        }
        else if (start_res == eServerBusy  ||  CTaskServer::IsInSoftShutdown()) {
            diag_msg.Flush();
            x_ReportOK("OK:SIZE=0, NEED_ABORT1\n");
            GetDiagCtx()->SetRequestStatus(eStatus_SyncAborted);
            // Old NC servers (those which used CNetCacheAPI instead of
            // CNCActiveHandler) always started to write blob data in SYNC_PUT
            // even when we responded to them NEED_ABORT. To avoid breaking the protocol
            // we need to read from them those fake blob writes.
            bool needs_fake = x_IsFlagSet(fReadExactBlobSize)  &&  m_CmdVersion == 0;
            if (start_res == eServerBusy)
                x_ResetFlags();
            if (needs_fake)
                return &CNCMessageHandler::x_ReadBlobSignature;
            return &CNCMessageHandler::x_FinishCommand;
        }
    }
 
    if (x_IsFlagSet(fNeedsSpaceAsPeer)
        && !CNCBlobStorage::AcceptWritesFromPeers()) {
        x_ReportError(eStatus_NoDiskSpace);
        return &CNCMessageHandler::x_FinishCommand;
    }
        
    if (!x_IsFlagSet(fNeedsBlobAccess) && !x_IsFlagSet(fNeedsBlobList)) {
        diag_msg.Flush();
        // if we do not need blob access
        return m_CmdProcessor;
    }
 
    if (((m_BlobPass.empty()  &&  m_AppSetup->pass_policy == eNCOnlyWithPass)
            ||  (!m_BlobPass.empty()  &&  m_AppSetup->pass_policy == eNCOnlyWithoutPass))
        &&  !x_IsFlagSet(fDoNotCheckPassword))
    {
        diag_msg.Flush();
        GetDiagCtx()->SetRequestStatus(eStatus_NotAllowed);
        // why substr?
        x_ReportError(GetMessageByStatus(eStatus_NotAllowed).substr(4));
        SRV_LOG(Warning, GetMessageByStatus(eStatus_NotAllowed));
        return &CNCMessageHandler::x_FinishCommand;
    }
 
    if (x_IsFlagSet(fCanGenerateKey)  &&  m_NCBlobKey.RawKey().empty()) {
        string raw_key;
//to test
//        bool old_ver = (CSrvTime::Current().CurSecs() % 2) != 0;
        bool old_ver = !x_IsHttpMode();
 
        CNCDistributionConf::GenerateBlobKey(
            !x_IsHttpMode() ? m_LocalPort : Uint4( CNCDistributionConf::GetSelfID()),
            raw_key, m_BlobSlot, m_TimeBucket, old_ver ? 1 : 3);
        m_NCBlobKey.Assign(raw_key);
        diag_msg.PrintParam("key", m_NCBlobKey.RawKey());
        diag_msg.PrintParam("gen_key", "1");
    }
    else if (!m_NCBlobKey.IsValid() && !x_IsFlagSet(fNeedsBlobList)) {
        diag_msg.Flush();
        x_ReportError(eStatus_NotFound);
        SRV_LOG(Warning, "Invalid blob key format: " << m_NCBlobKey.RawKey());
        return &CNCMessageHandler::x_FinishCommand;
    }
    else if (m_NCBlobKey.IsICacheKey()) {
        CNCDistributionConf::GetSlotByICacheKey(m_NCBlobKey, m_BlobSlot, m_TimeBucket);
    } else {
        CNCDistributionConf::GetSlotByRnd(m_NCBlobKey.GetRandomPart(), m_BlobSlot, m_TimeBucket);
    }
 
    m_BlobSize = 0;
    if (m_Slot == 0) {
        diag_msg.PrintParam("slot", m_BlobSlot);
    }
 
    if ((!CNCDistributionConf::IsServedLocally(m_BlobSlot)
            ||  !CNCServer::IsCachingComplete())
        &&  !x_IsFlagSet(fDoNotProxyToPeers)
        &&  !m_ForceLocal
        && CNCDistributionConf::CountServersForSlot(m_BlobSlot) != 0)
    {
        diag_msg.PrintParam("proxy", "1");
        diag_msg.Flush();
        x_GetCurSlotServers();
        return &CNCMessageHandler::x_ProxyToNextPeer;
    }
        
    if (!x_IsFlagSet(fNeedsBlobAccess)) {
        diag_msg.Flush();
        // if we do not need blob access
        return m_CmdProcessor;
    }
 
    // no blob access before caching is done
    if (!CNCServer::IsCachingComplete())
    {
        diag_msg.Flush();
        x_ReportError(eStatus_JustStarted);
        if (!x_IsHttpMode()) {
            x_ResetFlags();
        }
        return &CNCMessageHandler::x_FinishCommand;
    }
 
    diag_msg.Flush();
 
    if (!CNCServer::IsInitiallySynced()  &&  !m_ForceLocal
        &&  x_IsFlagSet(fUsesPeerSearch))
    {
        m_Quorum = 0;
    }
    if (x_IsFlagSet(fNeedsLowerPriority))
        SetPriority(CNCDistributionConf::GetSyncPriority());
    if ((x_IsFlagSet(fNeedsSpaceAsClient)
            &&  CNCBlobStorage::NeedStopWrite())
        ||  (x_IsFlagSet(fNeedsSpaceAsPeer)
            &&  !CNCBlobStorage::AcceptWritesFromPeers()))
    {
        if (CNCDistributionConf::CountServersForSlot(m_BlobSlot) != 0) {
            x_GetCurSlotServers();
            return &CNCMessageHandler::x_ProxyToNextPeer;
        }
        x_ReportError(eStatus_NoDiskSpace);
        return &CNCMessageHandler::x_FinishCommand;
    }
 
    x_LogCmdEvent("RequestMetaInfo");
    m_BlobAccess = CNCBlobStorage::GetBlobAccess(
                                    x_IsUserFlagSet(fNoCreate) ? eNCNone : m_ParsedCmd.command->extra.blob_access,
                                    m_NCBlobKey.PackedKey(), m_BlobPass, m_TimeBucket);
    m_BlobAccess->RequestMetaInfo(this);
    return &CNCMessageHandler::x_WaitForBlobAccess;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_ReadCommand(void)
{
    LOG_CURRENT_FUNCTION
    if (NeedToClose()  ||  CTaskServer::IsInSoftShutdown()) {
        if (CTaskServer::IsInShutdown())
            GetDiagCtx()->SetRequestStatus(eStatus_ShuttingDown);
        else
            GetDiagCtx()->SetRequestStatus(eStatus_CmdTimeout);
        return &CNCMessageHandler::x_SaveStatsAndClose;
    }
 
    CTempString cmd_line;
    if (x_IsHttpMode() && !m_PosponedCmd.empty()) {
        cmd_line = m_PosponedCmd;
    } else if (!ReadLine(&cmd_line)) {
        if (!HasError()  &&  CanHaveMoreRead())
            return NULL;
        if (IsReadDataAvailable())
            GetDiagCtx()->SetRequestStatus(eStatus_PrematureClose);
        return &CNCMessageHandler::x_SaveStatsAndClose;
    }
 
    if (!x_IsHttpMode()) {
        try {
            m_ParsedCmd = m_Parser.ParseCommand(cmd_line);
        }
        catch (CNSProtoParserException& ex) {
            SRV_LOG(Warning, "Error parsing command: " << ex);
            GetDiagCtx()->SetRequestStatus(eStatus_BadCmd);
            return &CNCMessageHandler::x_SaveStatsAndClose;
        }
    } else {
        m_ClientParams.erase("key");
        list<CTempString> arr;
        ncbi_NStr_Split(cmd_line, " ", arr);
        bool good = false;
        if (arr.size() >= 3) {
            CTempString arr_cmd(arr.front());
            CTempString arr_uri(*(++arr.begin()));
            CTempString arr_key;
            if (arr_cmd == "DELETE" ||
                arr_cmd == "GET"    ||
                arr_cmd == "HEAD"   ||
                arr_cmd == "POST"   ||
                arr_cmd == "PUT") {
                {
                    // eg, "/"  "/service"
                    list<CTempString> uri_parts;
                    ncbi_NStr_Split(arr_uri, "/", uri_parts);
                    if (uri_parts.size() > 0) {
                        if (arr_cmd != "POST") {
                            arr_key = uri_parts.back();
                            m_ClientParams["key"] = arr_key;
                            uri_parts.pop_back();
                        }
                        string service;
                        if (!uri_parts.empty())  {
                            service = NStr::Join(uri_parts, "/");
                        }
                        m_ClientParams["service"] = service;
                    }
                }
                m_ClientParams["cache"].clear();
 
                try {
                    m_ParsedCmd = m_Parser.ParseCommand(cmd_line);
                    good = true;
                }
                catch (CNSProtoParserException& ) {
                    GetDiagCtx()->SetRequestStatus(eStatus_BadCmd);
                }
            } else {
                GetDiagCtx()->SetRequestStatus(eStatus_NoImpl);
                SRV_LOG(Error, "Unrecognized command: " << cmd_line);
            }
        } else {
            SRV_LOG(Error, "Error parsing command: " << cmd_line);
            GetDiagCtx()->SetRequestStatus(eStatus_BadCmd);
        }
        m_PosponedCmd.clear();
        if (!good) {
            x_WriteHttpResponse();
            return &CNCMessageHandler::x_SaveStatsAndClose;
        }
    }
 
    const SCommandExtra& cmd_extra = m_ParsedCmd.command->extra;
    m_CmdProcessor = cmd_extra.processor;
    m_Flags        = cmd_extra.cmd_flags;
    CreateNewDiagCtx();
    try {
        x_AssignCmdParams();
    }
    catch (CStringException& ex) {
        ReleaseDiagCtx();
        SRV_LOG(Warning, "Error while parsing command '" << cmd_line
                         << "': " << ex);
        GetDiagCtx()->SetRequestStatus(eStatus_BadCmd);
        return &CNCMessageHandler::x_SaveStatsAndClose;
    }
    return &CNCMessageHandler::x_StartCommand;
}
 
void
CNCMessageHandler::x_GetCurSlotServers(void)
{
    LOG_CURRENT_FUNCTION
    CNCDistributionConf::GetServersForSlot(m_BlobSlot, m_CheckSrvs);
    m_SrvsIndex = 0;
    Uint4 main_srv_ip = 0;
    if (!m_NCBlobKey.IsICacheKey()) {
        main_srv_ip = CNCDistributionConf::GetMainSrvIP(m_NCBlobKey);
    }
    m_ThisServerIsMain = false;
    if (main_srv_ip != 0) {
        // Note: this check for "main" server for blob assumes that for each
        // blob slot only one NetCache instance processing it works on each
        // server. It will give false positive results if there are several
        // NC instances on the same server processing the same slot. But there's
        // no much sense in such setup, so it's pretty safe assumption.
        if (Uint4(CNCDistributionConf::GetSelfID() >> 32) == main_srv_ip
            // if NeedStopWrite, I cannot be sure that blob on this server is most recent
            // and need to ask other servers
            && !CNCBlobStorage::NeedStopWrite()
            // make sure this slot is served here
            && CNCDistributionConf::IsServedLocally(m_BlobSlot)
            ) {
            //m_ThisServerIsMain = true;
            m_ThisServerIsMain = 
                CNCBlobAccessor::HasPutSucceeded(m_NCBlobKey.PackedKey());
        }
        if (!m_ThisServerIsMain) {
            CWriteBackControl::AnotherServerMain();
            for (size_t i = 0; i < m_CheckSrvs.size(); ++i) {
                if (Uint4(m_CheckSrvs[i] >> 32) == main_srv_ip) {
                    Uint8 srv_id = m_CheckSrvs[i];
                    m_CheckSrvs.erase(m_CheckSrvs.begin() + i);
                    m_CheckSrvs.insert(m_CheckSrvs.begin(), srv_id);
                    break;
                }
            }
        }
    }
}
 
void
CNCMessageHandler::x_JournalBlobPutResult(int status, const string& blob_key, Uint2 blob_slot)
{
    if (CNCDistributionConf::CountServersForSlot(blob_slot) != 0) {
        if (status == eStatus_PrematureClose && status == eStatus_CmdTimeout) {
            CNCBlobAccessor::PutFailed(blob_key);
        } else {
            CNCBlobAccessor::PutSucceeded(blob_key);
        }
    }
}
 
void
CNCMessageHandler::x_ReportError( EHTTPStatus sts, bool eol /*= true*/)
{
    GetDiagCtx()->SetRequestStatus(sts);
    x_ReportError( GetMessageByStatus(sts), eol);
}
 
void
CNCMessageHandler::x_ReportError( const string& sts, bool eol /*= true*/)
{
// in HTTP mode we leave it to x_CleanCmdResources always
    if (!x_IsHttpMode()) {
        if (!x_IsFlagSet(fNoReplyOnFinish)) {
            x_SetFlag(fNoReplyOnFinish);
            WriteText(sts);
            if (eol) {
                WriteText("\n");
            }
        }
    }
}
 
CNCMessageHandler&
CNCMessageHandler::x_ReportOK(const string& sts)
{
    if (!x_IsHttpMode()) {
        if (!x_IsFlagSet(fNoReplyOnFinish)) {
            x_SetFlag(fNoReplyOnFinish);
            WriteText(sts);
        }
    }
    return *this;
}
 
void 
CNCMessageHandler::x_LogCmdEvent( const CTempString& evt)
{
    CSrvTime cmd_now = CSrvTime::Current();
    CSrvTime cmd_len = cmd_now;
    cmd_len -= m_CmdPrevTime;
    CSrvTime cmd_time = cmd_now;
    cmd_time -= m_CmdStartTime;
    m_CmdPrevTime = cmd_now;
    m_CmdLog.push_back( evt + ": " + NStr::NumericToString(cmd_len.AsUSec())  + "us ("
                                   + NStr::NumericToString(cmd_time.AsUSec()) + "us)");
}
 
void CNCMessageHandler::x_LogCmdLog(void)
{
    for( const string& l : m_CmdLog) {
        SRV_LOG(Warning, l);
    }
}
 
CNCMessageHandler::State
CNCMessageHandler::x_WaitForBlobAccess(void)
{
    LOG_CURRENT_FUNCTION
    if (!m_BlobAccess->IsMetaInfoReady())
        return NULL;
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_CloseCmdAndConn;
 
    if (!m_BlobAccess->IsAuthorized()  &&  !x_IsFlagSet(fDoNotCheckPassword)) {
        x_ReportError(eStatus_BadPassword);
        return &CNCMessageHandler::x_FinishCommand;
    }
 
// send COPY_UPD notification early, before blob data is received
//  another option is to send it in x_FinishReadingBlob,
//      after m_BlobAccess->Finalize(), before  CNCPeerControl::MirrorWrite
// sending the notification here we assume the risk that the blob will not be received correctly
// and the notification will create confusion only (it will be corrected by periodic sync).
    if (
        (m_Flags & eProxyBlobWrite) == eProxyBlobWrite // request from clients only
        && CNCDistributionConf::GetBlobUpdateHotline()
#if !USE_ALWAYS_COPY_UPD
        // if the blob exists here, it can exist on mirrors as well
        && (m_BlobAccess->IsBlobExists()  
        // if blob does not exist here, but it is created on another server, I should notify them
            || !CNCDistributionConf::IsThisServerKey(m_NCBlobKey))
#endif
        ) {
        CNCPeerControl::MirrorUpdate(m_NCBlobKey, m_BlobSlot, CSrvTime::Current().AsUSec());
    }
 
    if (!x_IsFlagSet(fUsesPeerSearch)) {
        x_LogCmdEvent("CmdProcessor");
        return m_CmdProcessor;
    }
 
    // All commands that have fUsesPeerSearch will operate on m_LatestExist and
    // m_LatestBlobSum, so we need to fill it here even if in next "if" we'll go
    // almost directly to m_CmdProcessor.
    bool is_exist = m_BlobAccess->IsBlobExists();
    bool is_valid = m_BlobAccess->IsValid();
    m_LatestExist = is_exist
                    &&  (x_IsFlagSet(fNoBlobVersionCheck)
                         ||  m_BlobAccess->GetCurBlobVersion() == m_BlobVersion);
    if (x_IsFlagSet(fDoNotProxyToPeers) || m_ForceLocal || !m_SearchOnRead) {
        m_LatestSrvId = CNCDistributionConf::GetSelfID();
    } else {
        m_LatestSrvId = is_valid ?
            (is_exist ? CNCDistributionConf::GetSelfID() : CNCDistributionConf::GetMainSrvId(m_NCBlobKey)) :
            m_BlobAccess->GetValidServer();
    }
    if (m_LatestSrvId != 0) {
        if (m_LatestExist) {
            m_LatestBlobSum->create_time   = m_BlobAccess->GetCurBlobCreateTime();
            m_LatestBlobSum->create_server = m_BlobAccess->GetCurCreateServer();
            m_LatestBlobSum->create_id     = m_BlobAccess->GetCurCreateId();
            m_LatestBlobSum->dead_time     = m_BlobAccess->GetCurBlobDeadTime();
            m_LatestBlobSum->expire        = m_BlobAccess->GetCurBlobExpire();
            m_LatestBlobSum->ver_expire    = m_BlobAccess->GetCurVerExpire();
        }
        if (x_IsFlagSet(fDoNotProxyToPeers)
            ||  m_ForceLocal
            ||  (m_Quorum == 1  &&  (m_LatestExist  ||  !m_SearchOnRead))
            ||  (m_LatestExist  &&  x_IsFlagSet(fPeerFindExistsOnly)))
        {
            return &CNCMessageHandler::x_ExecuteOnLatestSrvId;
        }
    }
 
    x_GetCurSlotServers();
    if (m_ThisServerIsMain
        &&  m_AppSetup->fast_on_main
        &&  CNCServer::IsInitiallySynced())
    {
        if (!is_valid) {
            m_CheckSrvs.clear();
            m_SrvsIndex = 0;
        }
        return &CNCMessageHandler::x_ExecuteOnLatestSrvId;
    }
    if (!is_valid) {
        Uint8 srv_id = m_BlobAccess->GetValidServer();
        if (srv_id) {
            for (size_t i = 0; i < m_CheckSrvs.size(); ++i) {
                if (m_CheckSrvs[i] == srv_id) {
                    m_CheckSrvs.erase(m_CheckSrvs.begin() + i);
                    m_CheckSrvs.insert(m_CheckSrvs.begin(), srv_id);
                    break;
                }
            }
        }
    }
 
    if (m_LatestExist  &&  m_Quorum != 0) {
        --m_Quorum;
    }
    x_LogCmdEvent("ReadMetaNextPeer");
    return &CNCMessageHandler::x_ReadMetaNextPeer;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_ReportBlobNotFound(void)
{
    LOG_CURRENT_FUNCTION
    x_SetFlag(fNoBlobAccessStats);
    x_ReportError(eStatus_NotFound, false);
    if (!x_IsHttpMode()) {
        if (m_AgeMax != 0 && m_BlobAccess->IsBlobExists()) {
            WriteText(", AGE=").WriteNumber(m_AgeCur);
            WriteText(", VER=").WriteNumber(m_BlobAccess->GetCurBlobVersion());
        }
        WriteText("\n");
    }
    return &CNCMessageHandler::x_FinishCommand;
}
 
void
CNCMessageHandler::x_ProlongBlobDeadTime(unsigned int add_time)
{
    LOG_CURRENT_FUNCTION
 
    CSrvTime cur_srv_time = CSrvTime::Current();
    Uint8 cur_time = cur_srv_time.AsUSec();
    int now = int(cur_srv_time.Sec());
    int new_expire = now + add_time;
    if (new_expire < now) {
        return;
    }
    int old_expire = m_BlobAccess->GetCurBlobExpire();
    if (!CNCServer::IsDebugMode() &&
        new_expire - old_expire < (int)m_AppSetup->ttl_unit) {
        return;
    }
 
    if (m_AppSetup->lifespan_ttl > 0) {
        int created = (int)(m_BlobAccess->GetCurBlobCreateTime()/kUSecsPerSecond);
        int retire = (int)(created + m_AppSetup->lifespan_ttl);
        if (retire > created) {
            new_expire = min(new_expire,retire);
        }
    }
 
    m_BlobAccess->SetCurBlobExpire(new_expire);
    if (m_BlobAccess->GetCurBlobSize() <= CNCDistributionConf::GetMaxBlobSizeSync()) {
        SNCSyncEvent* event = new SNCSyncEvent();
        event->blob_size = m_BlobAccess->GetCurBlobSize();
        event->event_type = eSyncProlong;
        event->key = m_NCBlobKey;
        event->orig_server = CNCDistributionConf::GetSelfID();
        event->orig_time = cur_time;
        CNCSyncLog::AddEvent(m_BlobSlot, event);
        CNCPeerControl::MirrorProlong(m_NCBlobKey, m_BlobSlot,
                                      event->orig_rec_no, cur_time, m_BlobAccess);
    }
}
 
void
CNCMessageHandler::x_ProlongVersionLife(void)
{
    LOG_CURRENT_FUNCTION
    CSrvTime cur_srv_time = CSrvTime::Current();
    Uint8 cur_time = cur_srv_time.AsUSec();
    int new_expire = int(cur_srv_time.Sec()) + m_BlobAccess->GetCurVersionTTL();
    int old_expire = m_BlobAccess->GetCurVerExpire();
    if (!CNCServer::IsDebugMode()  &&  new_expire - old_expire < m_AppSetup->ttl_unit)
        return;
 
    m_BlobAccess->SetCurVerExpire(new_expire);
    if (m_BlobAccess->GetCurBlobSize() <= CNCDistributionConf::GetMaxBlobSizeSync()) {
        SNCSyncEvent* event = new SNCSyncEvent();
        event->blob_size = m_BlobAccess->GetCurBlobSize();
        event->event_type = eSyncProlong;
        event->key = m_NCBlobKey;
        event->orig_server = CNCDistributionConf::GetSelfID();
        event->orig_time = cur_time;
        CNCSyncLog::AddEvent(m_BlobSlot, event);
        CNCPeerControl::MirrorProlong(m_NCBlobKey, m_BlobSlot,
                                      event->orig_rec_no, cur_time, m_BlobAccess);
    }
}
 
void
CNCMessageHandler::x_CleanCmdResources(void)
{
    LOG_CURRENT_FUNCTION
    int cmd_status = GetDiagCtx()->GetRequestStatus();
    bool print_size = false;
    Uint8 written_size = 0;
    ENCAccessType access_type = eNCCopyCreate;
    if (m_write_event) {
        delete m_write_event;
        m_write_event = NULL;
    }
    if (m_BlobAccess) {
        access_type = m_BlobAccess->GetAccessType();
        if (m_BlobAccess->IsBlobExists()  &&  !x_IsFlagSet(fNoBlobAccessStats)) {
            print_size = true;
            if (access_type == eNCRead  ||  access_type == eNCReadData)
                m_BlobSize = m_BlobAccess->GetCurBlobSize();
            else
                m_BlobSize = m_BlobAccess->GetNewBlobSize();
            if (access_type == eNCReadData)
                GetDiagCtx()->SetBytesWr(m_BlobAccess->GetSizeRead());
            else if (access_type == eNCCreate  ||  access_type == eNCCopyCreate)
                GetDiagCtx()->SetBytesRd(m_BlobSize);
        }
        else if (access_type == eNCCreate) {
            written_size = m_BlobAccess->GetNewBlobSize();
        }
        m_BlobAccess->Release();
        m_BlobAccess = NULL;
    }
    else if (m_ActiveHub) {
        switch (m_ParsedCmd.command->extra.proxy_cmd) {
        case eProxyRead:
        case eProxyReadLast:
            print_size = true;
            m_BlobSize = m_ActiveHub->GetHandler()->GetSizeRd();
            GetDiagCtx()->SetBytesWr(m_BlobSize);
            break;
        case eProxyWrite:
            print_size = true;
            m_BlobSize = m_ActiveHub->GetHandler()->GetSizeWr();
            GetDiagCtx()->SetBytesRd(m_BlobSize);
            break;
        default:
            break;
        }
        m_ActiveHub->GetHandler()->ResetSizeRdWr();
    }
 
    if (x_IsFlagSet(fRunsInStartedSync)) {
        if (x_IsCmdSucceeded(cmd_status)  ||  x_IsFlagSet(fSyncCmdSuccessful)) {
            CNCPeriodicSync::SyncCommandFinished(m_SrvId, m_Slot, m_SyncId);
        } else {
            CNCPeriodicSync::Cancel(m_SrvId, m_Slot, m_SyncId);
        }
    }
    if (!x_IsFlagSet(fNoReplyOnFinish)) {
        if (!x_IsHttpMode()) {
            if (x_IsCmdSucceeded(cmd_status)) {
                WriteText("OK:\n");
            } else {
                WriteText(GetMessageByStatus(EHTTPStatus(cmd_status))).WriteText("\n");
            }
        } else {
            x_WriteHttpResponse();
        }
    }
    Flush();
 
    if (m_ActiveHub) {
        m_ActiveHub->Release();
        m_ActiveHub = NULL;
    }
    m_CheckSrvs.clear();
    m_SrvsIndex = 0;
    m_ChunkLen = 0;
    m_LastPeerError.clear();
 
    CSrvTime cmd_len = CSrvTime::Current();
    cmd_len -= m_CmdStartTime;
    Uint8 len_usec = cmd_len.AsUSec();
    if (IsLongCommand(len_usec)) {
        x_LogCmdLog();
    }
 
    if (print_size  &&  (x_IsCmdSucceeded(cmd_status)  ||  m_BlobSize != 0))
        CSrvDiagMsg().PrintExtra().PrintParam("blob_size", m_BlobSize);
    CSrvDiagMsg().StopRequest();
 
//    CSrvTime cmd_len = CSrvTime::Current();
//    cmd_len -= m_CmdStartTime;
//    Uint8 len_usec = cmd_len.AsUSec();
    CNCStat::CmdFinished(m_ParsedCmd.command->cmd, len_usec, cmd_status);
    if (m_Flags & fComesFromClient) {
        if (access_type == eNCCreate) {
            if (print_size  &&  x_IsCmdSucceeded(cmd_status))
                CNCStat::ClientBlobWrite(m_BlobSize, len_usec);
            else
                CNCStat::ClientBlobRollback(written_size);
        }
        else if (access_type == eNCReadData) {
            if (print_size)
                CNCStat::ClientBlobRead(m_BlobSize, len_usec);
        }
    }
    ++m_CntCmds;
 
    if (x_IsFlagSet(fNeedsLowerPriority))
        SetPriority(GetDefaultTaskPriority());
 
    m_SendBuff.reset();
    ReleaseDiagCtx();
 
    m_PosponedCmd.clear();
}
 
CNCMessageHandler::State
CNCMessageHandler::x_FinishCommand(void)
{
    LOG_CURRENT_FUNCTION
    int status = GetDiagCtx()->GetRequestStatus();
    if (x_IsFlagSet(eBlobPut) && m_NCBlobKey.IsICacheKey()) {
        x_JournalBlobPutResult(status, m_NCBlobKey.PackedKey(), m_BlobSlot);
    }
    if (x_IsFlagSet(fCursedPUT2Cmd))
        return &CNCMessageHandler::x_CloseCmdAndConn;
 
    x_CleanCmdResources();
    SetState(&CNCMessageHandler::x_ReadCommand);
    SetRunnable();
    return NULL;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_StartReadingBlob(void)
{
    LOG_CURRENT_FUNCTION
    // Flushing the initial response line that client should receive before it
    // will start writing blob data.
    Flush();
    m_BlobSize = 0;
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_FinishCommand;
    else
        return &CNCMessageHandler::x_ReadBlobSignature;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_FinishReadingBlob(void)
{
    LOG_CURRENT_FUNCTION
    x_LogCmdEvent("FinishReadingBlob");
    bool fail = false, keep_conn = !x_IsFlagSet(fNoReplyOnFinish);
    string errmsg;
    if (x_IsFlagSet(fReadExactBlobSize)  &&  m_BlobSize != m_Size) {
        fail = true;
        x_ReportError(eStatus_CondFailed);
        SRV_LOG(Error, "Wrong data for blob size " << m_Size
                        << " (received " << m_BlobSize << " bytes)");
    }
    else if (m_BlobSize > CNCBlobStorage::GetMaxBlobSizeStore()) {
        fail = true;
        x_ReportError(eStatus_BlobTooBig);
        SRV_LOG(Error, "Blob size exceeds the allowed maximum of "
                        << CNCBlobStorage::GetMaxBlobSizeStore()
                        << " (received " << m_BlobSize << " bytes)");
    }
    if (fail) {
        if (keep_conn) {
            return &CNCMessageHandler::x_FinishCommand;
        } else {
            return &CNCMessageHandler::x_CloseCmdAndConn;
        }
    }
 
    if (!x_IsCmdSucceeded(GetDiagCtx()->GetRequestStatus()))
        return &CNCMessageHandler::x_FinishCommand;
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_CloseCmdAndConn;
 
    // Fill all new event data but not add it to CNCSyncLog until we execute
    // m_BlobAccess->Finalize().
    SNCSyncEvent* write_event = new SNCSyncEvent();
    write_event->blob_size = m_BlobSize;
    write_event->event_type = eSyncWrite;
    write_event->key = m_NCBlobKey;
    if (x_IsFlagSet(fCopyLogEvent)) {
        write_event->orig_time = m_BlobAccess->GetNewBlobCreateTime();
        write_event->orig_server = m_BlobAccess->GetNewCreateServer();
        write_event->orig_rec_no = m_OrigRecNo;
    }
    else {
        CSrvTime cur_srv_time = CSrvTime::Current();
        Uint8 cur_time = cur_srv_time.AsUSec();
        int cur_secs = int(cur_srv_time.Sec());
        m_BlobAccess->SetBlobCreateTime(cur_time);
        if (x_IsUserFlagSet(fNoProlong) && m_BlobAccess->IsBlobExists() && !m_BlobAccess->IsCurBlobExpired()) {
            m_BlobAccess->SetNewBlobExpire(m_BlobAccess->GetCurBlobExpire());
            m_BlobAccess->SetNewVerExpire(m_BlobAccess->GetCurVerExpire());
        } else {
            if (m_BlobAccess->GetNewBlobExpire() == 0) {
                m_BlobAccess->SetNewBlobExpire(cur_secs + m_BlobAccess->GetNewBlobTTL());
            }
            m_BlobAccess->SetNewVerExpire(cur_secs + m_BlobAccess->GetNewVersionTTL());
        }
        m_BlobAccess->SetCreateServer(CNCDistributionConf::GetSelfID(),
                                      CNCBlobStorage::GetNewBlobId());
        write_event->orig_server = CNCDistributionConf::GetSelfID();
        write_event->orig_time = cur_time;
    }
 
    x_LogCmdEvent("Finalize");
    m_BlobAccess->Finalize();
    if (m_BlobAccess->HasError()) {
        delete write_event;
        GetDiagCtx()->SetRequestStatus(eStatus_ServerError);
        if (x_IsFlagSet(fNoReplyOnFinish))
            return &CNCMessageHandler::x_CloseCmdAndConn;
 
        x_ReportError("ERR:Error while reading blob");
        return &CNCMessageHandler::x_FinishCommand;
    }
 
    m_MirrorsDone.clear();
    if (!x_IsFlagSet(fCopyLogEvent)) {
        if (m_BlobAccess->GetNewBlobSize() <= CNCDistributionConf::GetMaxBlobSizeSync()) {
#if 0
            if (m_BlobAccess->IsCurBlobExpired()) {
                delete write_event;
                CNCPeerControl::MirrorRemove(m_NCBlobKey, m_BlobSlot, m_BlobAccess->GetNewBlobCreateTime()-1);
                return &CNCMessageHandler::x_FinishCommand;
            }
#endif
            // If fCopyLogEvent is not set then this blob comes from client and
            // thus we need to check quorum value before answering to client.
            // If fCopyLogEvent is set then this write comes from other server
            // and we don't care about quorum in this case.
            if (m_Quorum != 1) {
                if (m_Quorum != 0)
                    --m_Quorum;
                x_GetCurSlotServers();
                m_write_event = write_event;
// probably, this was made intentionally, but now it looks wrong
// let us respect quorum setting always
//                if (!m_ThisServerIsMain  ||  !m_AppSetup->fast_on_main)
                    return &CNCMessageHandler::x_PutToNextPeer;
            }
            m_OrigRecNo = CNCSyncLog::AddEvent(m_BlobSlot, write_event);
            CNCPeerControl::MirrorWrite(m_NCBlobKey, m_BlobSlot,
                                        m_OrigRecNo, m_BlobAccess->GetNewBlobSize(), m_MirrorsDone);
        } else {
            if (CNCDistributionConf::GetWarnBlobSizeSync()) {
                SRV_LOG(Warning, "Received blob is too big and will not be mirrored:"
                    << " blob key:"     << m_NCBlobKey.RawKey()
                    << " blob size: "   << m_BlobAccess->GetNewBlobSize()
                    << " max allowed: " << CNCDistributionConf::GetMaxBlobSizeSync());
            }
            if (CNCDistributionConf::IsThisServerKey(m_NCBlobKey)) {
                CNCPeerControl::MirrorRemove(m_NCBlobKey, m_BlobSlot, m_BlobAccess->GetNewBlobCreateTime()-1);
            } else {
                Uint8 srvId = CNCDistributionConf::GetMainSrvId(m_NCBlobKey);
                if (srvId != 0) {
                    m_CheckSrvs.push_back(srvId);
                    m_SrvsIndex = 0;
                    return &CNCMessageHandler::x_PutToNextPeer;
                }
            }
            delete write_event;
        }
    }
    else if (m_OrigRecNo != 0) {
        CNCSyncLog::AddEvent(m_BlobSlot, write_event);
    }
    else {
        // m_OrigRecNo can be 0 if blob comes from another server as a result
        // of synchronization by blob lists. In this case there's no event to
        // link to and thus we don't need to add event to our sync log.
        delete write_event;
    }
 
    return &CNCMessageHandler::x_FinishCommand;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_CloseOnPeerError(void)
{
    LOG_CURRENT_FUNCTION
    SRV_LOG(Warning, "Error executing command on peer "
        << m_ActiveHub->GetFullPeerName() << ", peer says: "
        << m_ActiveHub->GetErrMsg());
    GetDiagCtx()->SetRequestStatus(eStatus_PeerError);
    return &CNCMessageHandler::x_CloseCmdAndConn;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_ReadBlobSignature(void)
{
    LOG_CURRENT_FUNCTION
    if (x_IsHttpMode()) {
        return &CNCMessageHandler::x_ReadBlobChunkLength;
    }
 
    Uint4 sig = 0;
    bool has_sig = ReadNumber(&sig);
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_CloseCmdAndConn;
    if (!has_sig)
        return NULL;
 
    x_LogCmdEvent("ReadBlobSignature");
    if (sig == 0x04030201) {
        x_SetFlag(fSwapLengthBytes);
        return &CNCMessageHandler::x_ReadBlobChunkLength;
    }
    if (sig == 0x01020304) {
        x_UnsetFlag(fSwapLengthBytes);
        return &CNCMessageHandler::x_ReadBlobChunkLength;
    }
 
    GetDiagCtx()->SetRequestStatus(eStatus_BadCmd);
    SRV_LOG(Error, "Cannot determine the byte order. Got: "
                   << NStr::UIntToString(sig, 0, 16));
    return &CNCMessageHandler::x_CloseCmdAndConn;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_ReadBlobChunkLength(void)
{
    LOG_CURRENT_FUNCTION
    if (m_ActiveHub) {
        if (ProxyHadError()) {
            // If we were proxying blob data from client to another server and some
            // error occurred protocol doesn't allow us to anything else but
            // close both connections - to client and to other NC server.
            CSrvSocketTask* active_sock = m_ActiveHub->GetHandler()->GetSocket();
            if (active_sock  &&  active_sock->HasError())
                return &CNCMessageHandler::x_CloseOnPeerError;
            else
                return &CNCMessageHandler::x_CloseCmdAndConn;
        }
        ENCClientHubStatus status = m_ActiveHub->GetStatus();
        if (status == eNCHubError)
            return &CNCMessageHandler::x_CloseOnPeerError;
        else if (status != eNCHubCmdInProgress) {
            SRV_FATAL("Unexpected client status: " << status);
        }
 
        if (m_ChunkLen != 0) {
            CNCStat::ClientDataWrite(m_ChunkLen);
            CNCStat::PeerDataRead(m_ChunkLen);
        }
    }
 
    bool has_chunklen = true;
    if (x_IsFlagSet(fSkipBlobEOF)  &&  m_BlobSize == m_Size) {
        // Workaround for old STRS
        m_ChunkLen = 0xFFFFFFFF;
    }
    else if (x_IsHttpMode()) {
        m_ChunkLen = m_Size;
        has_chunklen = m_ChunkLen != 0;
    }
    else {
        has_chunklen = ReadNumber(&m_ChunkLen);
        if (!has_chunklen  &&  !CanHaveMoreRead()  &&  x_IsFlagSet(fCursedPUT2Cmd)) {
            return &CNCMessageHandler::x_FinishReadingBlob;
        }
    }
 
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_CloseCmdAndConn;
    if (!has_chunklen)
        return NULL;
 
    if (x_IsFlagSet(fSwapLengthBytes))
        m_ChunkLen = CByteSwap::GetInt4((const unsigned char*)&m_ChunkLen);
    if (m_ChunkLen == 0xFFFFFFFF) {
        if (m_ActiveHub) {
            // Data transfer is finished, CNCActiveHandler will wait for response
            // from other NC server and wake us up.
            m_ActiveHub->GetHandler()->SetRunnable();
            return &CNCMessageHandler::x_WaitForPeerAnswer;
        }
        return &CNCMessageHandler::x_FinishReadingBlob;
    }
 
    if (!m_BlobAccess  &&  !m_ActiveHub) {
        // We can be here only when expecting fake start of blob writing from old
        // NC server, but for some reason we got non-EOF chunk length.
        GetDiagCtx()->SetRequestStatus(eStatus_BadCmd);
        SRV_LOG(Critical, "Received non-EOF chunk len from peer "
            << m_SrvId << ": " << m_ChunkLen);
        return &CNCMessageHandler::x_CloseCmdAndConn;
    }
 
    if (x_IsFlagSet(fComesFromClient)) {
        if (m_BlobSize > CNCBlobStorage::GetMaxBlobSizeStore()) {
            RunAfter(1);
            return NULL;
        }
// this is potentially dangerous, because
// if we close connection here, in the middle of transmission,
// client might have no chance (most likely) to receive our error message
// and might want to retry thinking it was bad luck
        if (x_IsFlagSet(fReadExactBlobSize)  &&  (m_BlobSize + m_ChunkLen) > m_Size) {
            x_ReportError(eStatus_CondFailed);
            SRV_LOG(Error, "Too much data for blob size " << m_Size
                            << " (received at least "
                            << (m_BlobSize + m_ChunkLen) << " bytes)");
            return &CNCMessageHandler::x_CloseCmdAndConn;
        }
        if ((m_BlobSize + m_ChunkLen) > CNCBlobStorage::GetMaxBlobSizeStore()) {
            x_ReportError(eStatus_BlobTooBig);
            if (x_IsHttpMode()) {
                x_WriteHttpHeader(eStatus_BlobTooBig, 0, false);
            }
            SRV_LOG(Error, "Blob size exceeds the allowed maximum of "
                            << CNCBlobStorage::GetMaxBlobSizeStore()
                            << " (received " << m_BlobSize
                            << ", next chunk " << m_ChunkLen << " bytes)");
            // I am not going to read it anyway
            m_BlobSize += m_ChunkLen;
            Flush();
            RunAfter(1);
            return NULL;
//            return &CNCMessageHandler::x_CloseCmdAndConn;
        }
    }
 
    if (m_ActiveHub) {
        CSrvSocketTask* active_sock = m_ActiveHub->GetHandler()->GetSocket();
        active_sock->WriteData(&m_ChunkLen, sizeof(m_ChunkLen));
        m_ActiveHub->GetHandler()->AddSizeWr(m_ChunkLen);
        StartProxyTo(active_sock, m_ChunkLen);
        if (IsProxyInProgress())
            return NULL;
        else
            return &CNCMessageHandler::x_ReadBlobChunkLength;
    }
 
    return &CNCMessageHandler::x_ReadBlobChunk;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_ReadBlobChunk(void)
{
    LOG_CURRENT_FUNCTION
// there are too many of them
//    x_LogCmdEvent("ReadBlobChunk");
    while (m_ChunkLen != 0) {
        Uint4 read_len = Uint4(m_BlobAccess->GetWriteMemSize());
        if (m_BlobAccess->HasError()) {
            GetDiagCtx()->SetRequestStatus(eStatus_ServerError);
            if (!x_IsFlagSet(fNoReplyOnFinish) && !x_IsHttpMode()) {
                x_ReportError("ERR:Server error");
                Flush();
            }
            SetState(&CNCMessageHandler::x_CloseCmdAndConn);
            SetRunnable();
            return NULL;
        }
        if (read_len == 0)
            return NULL;
        if (read_len > m_ChunkLen)
            read_len = m_ChunkLen;
 
        Uint4 n_read = Uint4(Read(m_BlobAccess->GetWriteMemPtr(), read_len));
        if (n_read != 0) {
            if (m_Flags & fComesFromClient)
                CNCStat::ClientDataWrite(n_read);
            else
                CNCStat::PeerDataWrite(n_read);
        }
        if (NeedEarlyClose())
            return &CNCMessageHandler::x_CloseCmdAndConn;
        if (n_read == 0)
            return NULL;
 
        m_BlobAccess->MoveWritePos(n_read);
        m_ChunkLen -= n_read;
        m_BlobSize += n_read;
    }
    return &CNCMessageHandler::x_ReadBlobChunkLength;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_WriteBlobData(void)
{
    LOG_CURRENT_FUNCTION
    x_LogCmdEvent("WriteBlobData");
    while (m_Size != 0) {
        if (m_BlobAccess->GetPosition() == m_BlobAccess->GetCurBlobSize())
            return &CNCMessageHandler::x_FinishCommand;
 
        Uint4 want_read = m_BlobAccess->GetReadMemSize();
        if (m_BlobAccess->HasError()) {
            GetDiagCtx()->SetRequestStatus(eStatus_ServerError);
            return &CNCMessageHandler::x_CloseCmdAndConn;
        }
        if (m_Size != Uint8(-1)  &&  m_Size < want_read)
            want_read = Uint4(m_Size);
 
        Uint4 n_written = Uint4(Write(m_BlobAccess->GetReadMemPtr(), want_read));
//        x_LogCmdEvent("Write");
        if (n_written != 0) {
            if (m_Flags & fComesFromClient)
                CNCStat::ClientDataRead(n_written);
            else
                CNCStat::PeerDataRead(n_written);
            m_BlobAccess->MoveReadPos(n_written);
            if (m_Size != Uint8(-1))
                m_Size -= n_written;
        }
        if (NeedEarlyClose())
            return &CNCMessageHandler::x_CloseCmdAndConn;
        if (n_written == 0)
            return NULL;
    }
    return &CNCMessageHandler::x_FinishCommand;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_WriteSendBuff(void)
{
    LOG_CURRENT_FUNCTION
    while (m_SendPos != m_SendBuff->size()) {
        size_t n_written = Write(m_SendBuff->data() + m_SendPos,
                                 m_SendBuff->size() - m_SendPos);
 
        if (NeedEarlyClose())
            return &CNCMessageHandler::x_CloseCmdAndConn;
        if (n_written == 0)
            return NULL;
 
        m_SendPos += n_written;
    }
    if (strcmp(m_ParsedCmd.command->cmd, "SYNC_START") == 0) {
        return &CNCMessageHandler::x_WriteSyncStartExtra;
    }
    ENCProxyCmd proxy_cmd = m_ParsedCmd.command->extra.proxy_cmd;
    if (proxy_cmd == eProxyGetBList2) {
        Flush();
        return &CNCMessageHandler::x_DoCmd_GetBListNext;
    }
    return &CNCMessageHandler::x_FinishCommand;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_WriteSyncStartExtra(void)
{
    LOG_CURRENT_FUNCTION
    WriteText("PURGE:\n");
    WriteText(CNCBlobAccessor::GetPurgeData()). WriteText(";\n");
    return &CNCMessageHandler::x_FinishCommand;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_ProxyToNextPeer(void)
{
    LOG_CURRENT_FUNCTION
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_CloseCmdAndConn;
    x_LogCmdEvent("ProxyToNextPeer");
    if (m_SrvsIndex < m_CheckSrvs.size()) {
        Uint8 srv_id = m_CheckSrvs[m_SrvsIndex++];
        if (m_ActiveHub) {
            SRV_FATAL("Previous client not released");
        }
        m_ActiveHub = CNCActiveClientHub::Create(srv_id, this);
        return &CNCMessageHandler::x_SendCmdAsProxy;
    }
 
    // Either there's no servers to execute this command on or all servers were
    // tried and some error was the result from all of them.
    SRV_LOG(Warning, "Got error on all peer servers, LastPeerError is " << m_LastPeerError);
    if (m_LastPeerError.empty())
        m_LastPeerError = "ERR:Cannot execute command on peer servers";
    GetDiagCtx()->SetRequestStatus( GetStatusByMessage(m_LastPeerError, eStatus_PeerError));
    x_ReportError(m_LastPeerError);
    return &CNCMessageHandler::x_FinishCommand;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_SendCmdAsProxy(void)
{
    LOG_CURRENT_FUNCTION
    ENCClientHubStatus status = m_ActiveHub->GetStatus();
    if (status == eNCHubWaitForConn)
        return NULL;
    x_LogCmdEvent("SendCmdAsProxy");
    ENCProxyCmd proxy_cmd = m_ParsedCmd.command->extra.proxy_cmd;
    CNCActiveHandler* pHandler = m_ActiveHub->GetHandler();
    if (status == eNCHubError ||
        status == eNCHubSuccess ||
        (proxy_cmd == eProxyGetBList && !pHandler->GetPeer()->AcceptsBList()) ||
        (proxy_cmd == eProxyGetBList2 && !pHandler->GetPeer()->AcceptsBList2()) ||
        !pHandler->GetPeer()->AcceptsBlobKey(m_NCBlobKey)
       ) {
        m_LastPeerError = m_ActiveHub->GetErrMsg();
        m_ActiveHub->Release();
        m_ActiveHub = NULL;
        return &CNCMessageHandler::x_ProxyToNextPeer;
    }
    if (status != eNCHubConnReady) {
        SRV_FATAL("Unexpected client status: " << status);
    }
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_CloseCmdAndConn;
 
 
    switch (proxy_cmd) {
    case eProxyRead:
        pHandler->ProxyRead(GetDiagCtx(), m_NCBlobKey, m_RawBlobPass,
                                             m_BlobVersion, m_StartPos, m_Size,
                                             m_Quorum, m_SearchOnRead, m_ForceLocal,
                                             m_AgeMax);
        break;
    case eProxyWrite:
        pHandler->ProxyWrite(GetDiagCtx(), m_NCBlobKey, m_RawBlobPass,
                                              m_BlobVersion, m_BlobTTL, m_Quorum, m_UserFlags);
        // The only place that needs to go further to a different state.
        return &CNCMessageHandler::x_WriteInitWriteResponse;
    case eProxyHasBlob:
        pHandler->ProxyHasBlob(GetDiagCtx(), m_NCBlobKey, m_RawBlobPass,
                                                m_Quorum);
        break;
    case eProxyGetSize:
        pHandler->ProxyGetSize(GetDiagCtx(), m_NCBlobKey, m_RawBlobPass,
                                                m_BlobVersion, m_Quorum,
                                                m_SearchOnRead, m_ForceLocal);
        break;
    case eProxyReadLast:
        pHandler->ProxyReadLast(GetDiagCtx(), m_NCBlobKey, m_RawBlobPass,
                                                 m_StartPos, m_Size, m_Quorum,
                                                 m_SearchOnRead, m_ForceLocal, m_AgeMax);
        break;
    case eProxySetValid:
        pHandler->ProxySetValid(GetDiagCtx(), m_NCBlobKey, m_RawBlobPass,
                                                 m_BlobVersion);
        break;
    case eProxyRemove:
        pHandler->ProxyRemove(GetDiagCtx(), m_NCBlobKey, m_RawBlobPass,
                                               m_BlobVersion, m_Quorum);
        break;
    case eProxyGetMeta:
        pHandler->ProxyGetMeta(GetDiagCtx(), m_NCBlobKey,
                                                m_Quorum, m_ForceLocal, m_HttpMode);
        break;
    case eProxyProlong:
        pHandler->ProxyProlong(GetDiagCtx(), m_NCBlobKey, m_RawBlobPass,
                                                m_BlobTTL, m_Quorum,
                                                m_SearchOnRead, m_ForceLocal);
        break;
    case eProxyGetBList:
        pHandler->ProxyBList(GetDiagCtx(), m_NCBlobKey, m_ForceLocal, nullptr);
        break;
    case eProxyGetBList2:
        pHandler->ProxyBList(GetDiagCtx(), m_NCBlobKey, m_ForceLocal, m_BlobFilter);
        break;
    default:
        SRV_FATAL("Unsupported command: " << m_ParsedCmd.command->extra.proxy_cmd);
    }
 
    return &CNCMessageHandler::x_WaitForPeerAnswer;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_WaitForPeerAnswer(void)
{
    LOG_CURRENT_FUNCTION
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_CloseCmdAndConn;
    ENCClientHubStatus status = m_ActiveHub->GetStatus();
    if (status == eNCHubCmdInProgress)
        return NULL;
 
    x_LogCmdEvent("WaitForPeerAnswer");
    if (status == eNCHubError) {
        if (m_ActiveHub->GetHandler()->GotClientResponse())
            return &CNCMessageHandler::x_CloseOnPeerError;
 
        m_LastPeerError = m_ActiveHub->GetErrMsg();
        SRV_LOG(Warning, "Error executing command on peer "
            << m_ActiveHub->GetFullPeerName() << ", peer says: "  << m_LastPeerError);
        m_ActiveHub->Release();
        m_ActiveHub = NULL;
        return &CNCMessageHandler::x_ProxyToNextPeer;
    }
    if (status != eNCHubSuccess) {
        SRV_FATAL("Unexpected client status: " << status);
    }
    ENCProxyCmd proxy_cmd = m_ParsedCmd.command->extra.proxy_cmd;
    if (proxy_cmd == eProxyGetBList2) {
        return &CNCMessageHandler::x_DoCmd_GetBListNext;
    }
 
    const string& err_msg = m_ActiveHub->GetErrMsg();
    EHTTPStatus rst = GetStatusByMessage(err_msg, eStatus_OK);
    if (rst != eStatus_OK) {
        GetDiagCtx()->SetRequestStatus(rst);
    }
    if (!err_msg.empty()) {
        x_ReportError(err_msg);
    }
    return &CNCMessageHandler::x_FinishCommand;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_ReadMetaNextPeer(void)
{
    LOG_CURRENT_FUNCTION
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_CloseCmdAndConn;
    if (m_SrvsIndex >= m_CheckSrvs.size())
        return &CNCMessageHandler::x_ExecuteOnLatestSrvId;
 
    Uint8 srv_id = m_CheckSrvs[m_SrvsIndex++];
    if (m_ActiveHub) {
        SRV_FATAL("Previous client not released");
    }
    m_ActiveHub = CNCActiveClientHub::Create(srv_id, this);
    return &CNCMessageHandler::x_SendGetMetaCmd;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_SendGetMetaCmd(void)
{
    LOG_CURRENT_FUNCTION
    ENCClientHubStatus status = m_ActiveHub->GetStatus();
    if (status == eNCHubWaitForConn)
        return NULL;
    if (status == eNCHubError || status == eNCHubSuccess ||
        !m_ActiveHub->GetHandler()->GetPeer()->AcceptsBlobKey(m_NCBlobKey)) {
        m_LastPeerError = m_ActiveHub->GetErrMsg();
        m_ActiveHub->Release();
        m_ActiveHub = NULL;
        return &CNCMessageHandler::x_ReadMetaNextPeer;
    }
    if (status != eNCHubConnReady) {
        SRV_FATAL("Unexpected client status: " << status);
    }
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_CloseCmdAndConn;
 
    m_ActiveHub->GetHandler()->SearchMeta(GetDiagCtx(), m_NCBlobKey);
    return &CNCMessageHandler::x_ReadMetaResults;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_ReadMetaResults(void)
{
    LOG_CURRENT_FUNCTION
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_CloseCmdAndConn;
    ENCClientHubStatus status = m_ActiveHub->GetStatus();
    if (status == eNCHubCmdInProgress)
        return NULL;
    if (status == eNCHubError)
        goto results_processed;
    if (status != eNCHubSuccess) {
        SRV_FATAL("Unexpected client status: " << status);
    }
 
    CNCActiveHandler* handler;
    handler = m_ActiveHub->GetHandler();
    const SNCBlobSummary* cur_blob_sum;
    cur_blob_sum = &handler->GetBlobSummary();
    bool cur_exist;
    cur_exist = handler->IsBlobExists();
    if (!cur_exist  &&  !x_IsFlagSet(fPeerFindExistsOnly))
        goto results_processed;
 
    if (cur_exist  &&  x_IsFlagSet(fPeerFindExistsOnly)) {
        m_LatestExist = true;
        m_LatestSrvId = m_CheckSrvs[m_SrvsIndex - 1];
        goto meta_search_finished;
    }
    if (cur_exist  &&  (!m_LatestExist  ||  m_LatestBlobSum->isOlder(*cur_blob_sum)))
    {
        m_LatestExist = true;
        m_LatestSrvId = m_CheckSrvs[m_SrvsIndex - 1];
        *m_LatestBlobSum = *cur_blob_sum;
    }
    if (cur_blob_sum->size > CNCDistributionConf::GetMaxBlobSizeSync()) {
        m_Quorum = 1;
    }
    if (m_Quorum == 1)
        goto meta_search_finished;
    if (m_Quorum != 0)
        --m_Quorum;
 
results_processed:
 
    m_LastPeerError = m_ActiveHub->GetErrMsg();
    m_ActiveHub->Release();
    m_ActiveHub = NULL;
    return &CNCMessageHandler::x_ReadMetaNextPeer;
 
meta_search_finished:
 
    m_LastPeerError = m_ActiveHub->GetErrMsg();
    m_ActiveHub->Release();
    m_ActiveHub = NULL;
    m_CheckSrvs.clear();
    m_SrvsIndex = 0;
    return &CNCMessageHandler::x_ExecuteOnLatestSrvId;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_ExecuteOnLatestSrvId(void)
{
    LOG_CURRENT_FUNCTION
    x_LogCmdEvent("ExecuteOnLatestSrvId");
    if (m_LatestSrvId == 0) {
        m_LatestSrvId = CNCDistributionConf::GetSelfID();
    }
    if (m_LatestExist) {
        // if max age specified, check age
        if (m_AgeMax != 0 && m_BlobAccess->IsBlobExists()) {
            CSrvTime cur_srv_time = CSrvTime::Current();
            unsigned int vttl = m_BlobAccess->GetCurVersionTTL();
            Uint8 creation = vttl != 0 ?
// see x_ProlongVersionLife
                (m_LatestBlobSum->ver_expire - vttl) :
                (m_LatestBlobSum->create_time / kUSecsPerSecond);
            m_AgeCur =  Uint8(cur_srv_time.Sec()) - creation;
            if (m_AgeCur > m_AgeMax) {
                return &CNCMessageHandler::x_ReportBlobNotFound;
            }
        }
    }
    if (m_BlobAccess->IsPurged(m_NCBlobKey)) {
        if (x_IsFlagSet(fPeerFindExistsOnly)) {
            m_LatestExist = false;
            return m_CmdProcessor;
        }
#if 0
        return &CNCMessageHandler::x_ReportBlobNotFound;
#else
        x_ReportError( GetMessageByStatus(eStatus_NotFound));
        return &CNCMessageHandler::x_DoCmd_Remove;
#endif
    }
    if (x_IsFlagSet(fPeerFindExistsOnly) && m_LatestSrvId == CNCDistributionConf::GetSelfID()) {
        return m_CmdProcessor;
    }
    if (m_LatestSrvId == CNCDistributionConf::GetSelfID()) {
        if (m_LatestExist  && m_BlobAccess->IsBlobExists() && !m_BlobAccess->IsCurBlobExpired()) {
            return m_CmdProcessor;
        } else {
            return &CNCMessageHandler::x_ReportBlobNotFound;
        }
    }
 
    CSrvDiagMsg().PrintExtra().PrintParam("proxy", "1");
    // Changing parameters that will go to other server: that server have to
    // execute command locally, without quorum (quorum equal to 1), and without
    // searching on other servers.
    m_Quorum = 1;
    m_SearchOnRead = false;
    m_ForceLocal = true;
    m_CheckSrvs.push_back(m_LatestSrvId);
    x_LogCmdEvent("ProxyToNextPeer");
    return &CNCMessageHandler::x_ProxyToNextPeer;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_PutToNextPeer(void)
{
    LOG_CURRENT_FUNCTION
    x_LogCmdEvent("PutToNextPeer");
    if (m_SrvsIndex >= m_CheckSrvs.size()  ||  NeedEarlyClose())
        return &CNCMessageHandler::x_FinishCommand;
 
    Uint8 srv_id = m_CheckSrvs[m_SrvsIndex++];
    if (m_ActiveHub) {
        SRV_FATAL("Previous client not released");
    }
    if (CNCDistributionConf::GetSelfTrustLevel() < CNCPeerControl::Peer(srv_id)->GetTrustLevel()) {
        return &CNCMessageHandler::x_PutToNextPeer;
    }
    m_ActiveHub = CNCActiveClientHub::Create(srv_id, this);
    return &CNCMessageHandler::x_SendPutToPeerCmd;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_SendPutToPeerCmd(void)
{
    LOG_CURRENT_FUNCTION
    ENCClientHubStatus status = m_ActiveHub->GetStatus();
    if (status == eNCHubWaitForConn)
        return NULL;
    if (status == eNCHubError || status == eNCHubSuccess ||
        !m_ActiveHub->GetHandler()->GetPeer()->AcceptsBlobKey(m_NCBlobKey)) {
        m_MirrorsDone.push_back(m_CheckSrvs[m_SrvsIndex-1]);
        m_LastPeerError = m_ActiveHub->GetErrMsg();
        m_ActiveHub->Release();
        m_ActiveHub = NULL;
        return &CNCMessageHandler::x_PutToNextPeer;
    }
    if (status != eNCHubConnReady) {
        SRV_FATAL("Unexpected client status: " << status);
    }
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_FinishCommand;
 
    m_ActiveHub->GetHandler()->CopyPut(GetDiagCtx(), m_NCBlobKey, m_BlobSlot, m_OrigRecNo);
    return &CNCMessageHandler::x_ReadPutResults;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_ReadPutResults(void)
{
    LOG_CURRENT_FUNCTION
    x_LogCmdEvent("ReadPutResults");
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_FinishCommand;
    ENCClientHubStatus status = m_ActiveHub->GetStatus();
    if (status == eNCHubCmdInProgress)
        return NULL;
    if (status == eNCHubError)
        goto results_processed;
    if (status != eNCHubSuccess) {
        SRV_FATAL("Unexpected client status: " << status);
    }
 
    m_MirrorsDone.push_back(m_CheckSrvs[m_SrvsIndex-1]);
    if (m_Quorum == 1) {
        if (m_BlobAccess->GetNewBlobSize() <= CNCDistributionConf::GetMaxBlobSizeSync()) {
            if (m_write_event) {
                m_OrigRecNo = CNCSyncLog::AddEvent(m_BlobSlot, m_write_event);
                m_write_event = NULL;
            }
            CNCPeerControl::MirrorWrite(m_NCBlobKey, m_BlobSlot,
                                        m_OrigRecNo, m_BlobAccess->GetNewBlobSize(), 
                                        m_MirrorsDone);
        }
        return &CNCMessageHandler::x_FinishCommand;
    }
    if (m_Quorum != 0)
        --m_Quorum;
 
results_processed:
    m_ActiveHub->Release();
    m_ActiveHub = NULL;
    return &CNCMessageHandler::x_PutToNextPeer;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_PurgeToNextPeer(void)
{
    LOG_CURRENT_FUNCTION
    if (m_SrvsIndex >= m_CheckSrvs.size()  ||  NeedEarlyClose())
        return &CNCMessageHandler::x_FinishCommand;
 
    Uint8 srv_id = m_CheckSrvs[m_SrvsIndex++];
    if (m_ActiveHub) {
        SRV_FATAL("Previous client not released");
    }
    if (!m_NCBlobKey.RawKey().empty() && !CNCPeerControl::Peer(srv_id)->AcceptsPurge2()) {
        return &CNCMessageHandler::x_PurgeToNextPeer;
    }
    m_ActiveHub = CNCActiveClientHub::Create(srv_id, this);
    return &CNCMessageHandler::x_SendPurgeToPeerCmd;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_SendPurgeToPeerCmd(void)
{
    LOG_CURRENT_FUNCTION
    ENCClientHubStatus status = m_ActiveHub->GetStatus();
    if (status == eNCHubWaitForConn)
        return NULL;
    if (status == eNCHubError || status == eNCHubSuccess) {
        m_LastPeerError = m_ActiveHub->GetErrMsg();
        m_ActiveHub->Release();
        m_ActiveHub = NULL;
        return &CNCMessageHandler::x_PurgeToNextPeer;
    }
    if (status != eNCHubConnReady) {
        SRV_FATAL("Unexpected client status: " << status);
    }
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_FinishCommand;
 
    m_ActiveHub->GetHandler()->CopyPurge(GetDiagCtx(), m_NCBlobKey, m_CmdStartTime.AsUSec());
    return &CNCMessageHandler::x_ReadPurgeResults;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_ReadPurgeResults(void)
{
    LOG_CURRENT_FUNCTION
    if (NeedEarlyClose())
        return &CNCMessageHandler::x_FinishCommand;
    ENCClientHubStatus status = m_ActiveHub->GetStatus();
    if (status == eNCHubCmdInProgress)
        return NULL;
    if (status == eNCHubError)
        goto results_processed;
    if (status != eNCHubSuccess) {
        SRV_FATAL("Unexpected client status: " << status);
    }
 
results_processed:
    m_ActiveHub->Release();
    m_ActiveHub = NULL;
    return &CNCMessageHandler::x_PurgeToNextPeer;
}
 
 
inline unsigned int
CNCMessageHandler::x_GetBlobTTL(void)
{
    LOG_CURRENT_FUNCTION
    return m_BlobTTL != 0 ? min(m_BlobTTL,m_AppSetup->max_ttl) : m_AppSetup->blob_ttl;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_DoCmd_Health(void)
{
    LOG_CURRENT_FUNCTION
    const char* health_coeff = "1";
    if (CNCBlobStorage::NeedStopWrite()) {
        health_coeff = "0 (does not accept writes)";
    } else if (!CNCServer::IsCachingComplete()) {
        health_coeff = "0.1 (caching not finished)";
    } else if (CNCBlobStorage::IsDraining()) {
        health_coeff = "0.2 (draining)";
    } else if (!CNCServer::IsInitiallySynced()) {
        health_coeff = "0.5 (initial sync not finished)";
    } else if (CNCPeerControl::HasPeerInThrottle()) {
        health_coeff = "0.8 (some peers unaccessible)";
    }
    x_ReportOK("OK:HEALTH_COEFF=").WriteText(health_coeff).WriteText("\n");
    WriteText("OK:UP_TIME=").WriteNumber(CNCServer::GetUpTime()).WriteText("\n");
    WriteText("OK:CACHING_COMPLETE=").WriteText(CNCServer::IsCachingComplete()? "yes": "no").WriteText("\n");
    WriteText("OK:INITIALLY_SYNCED=").WriteText(CNCServer::IsInitiallySynced()? "yes": "no").WriteText("\n");
    Int8 free_space = CNCBlobStorage::GetDiskFree();
    Int8 allowed_size = CNCBlobStorage::GetAllowedDBSize(free_space);
    WriteText("OK:DISK_FREE=").WriteNumber(free_space).WriteText("\n");
    WriteText("OK:DISK_LIMIT=").WriteNumber(allowed_size).WriteText("\n");
    WriteText("OK:DISK_USED=").WriteNumber(CNCBlobStorage::GetDBSize()).WriteText("\n");
    WriteText("OK:DISK_LIMIT_ALERT=").WriteText(CNCBlobStorage::IsDBSizeAlert()? "yes": "no").WriteText("\n");
    WriteText("OK:N_DB_FILES=").WriteNumber(CNCBlobStorage::GetNDBFiles()).WriteText("\n");
    WriteText("OK:COPY_QUEUE_SIZE=").WriteNumber(CNCPeerControl::GetMirrorQueueSize()).WriteText("\n");
 
    const TNCPeerList& peers = CNCDistributionConf::GetPeers();
    ITERATE(TNCPeerList, it_peer, peers) {
//        WriteText("OK:QUEUE_SIZE_").WriteNumber(it_peer->first).WriteText("=").WriteNumber(CNCPeerControl::GetMirrorQueueSize(it_peer->first)).WriteText("\n");
        WriteText("OK:QUEUE_SIZE_").WriteText(it_peer->second).WriteText("=").WriteNumber(CNCPeerControl::GetMirrorQueueSize(it_peer->first)).WriteText("\n");
    }
    WriteText("OK:SYNC_LOG_SIZE=").WriteNumber(CNCSyncLog::GetLogSize()).WriteText("\n");
 
    WriteText("OK:END\n");
    return &CNCMessageHandler::x_FinishCommand;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_DoCmd_Shutdown(void)
{
    LOG_CURRENT_FUNCTION
    TNSProtoParams& param = m_ParsedCmd.params;
    if (param.find("drain") != param.end() && param["drain"] != "0") {
        CNCBlobStorage::SetDraining(true);
    } else if (param.find("reset") != param.end() && param["reset"] != "0") {
        CNCBlobStorage::AbandonDB();
        CTaskServer::RequestShutdown(eSrvFastShutdown);
    } else {
        CTaskServer::RequestShutdown(eSrvSlowShutdown);
    }
    return &CNCMessageHandler::x_FinishCommand;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_DoCmd_Version(void)
{
    LOG_CURRENT_FUNCTION
    x_ReportOK("OK:server_version="   NETCACHED_VERSION
              "&storage_version="  NETCACHED_STORAGE_VERSION
              "&protocol_version=" NETCACHED_PROTOCOL_VERSION
              "&build_date=" + NStr::URLEncode(NETCACHED_BUILD_DATE) +
              "&mirrored=" + (NStr::BoolToString( CNCDistributionConf::HasPeers())))
        .WriteText("\n");
    return &CNCMessageHandler::x_FinishCommand;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_DoCmd_GetConfig(void)
{
    LOG_CURRENT_FUNCTION
    TNSProtoParams& params = m_ParsedCmd.params;
    if (params.find("section") != params.end()) {
        string section(params["section"]);
        WriteText("{\"").WriteText(section).WriteText("\": {\n\"section\": \"");
        WriteText(section).WriteText("\"");
        if (section == "task_server") {
            CTaskServer::WriteSetup(*this);
        } else if (section == "netcache") {
            TStringMap client;
            if (params.find("port") != params.end()) {
                client["port"] = params["port"];
            }
            if (params.find("cache") != params.end()) {
                client["cache"] = params["cache"];
            }
            Flush();
            m_SendBuff.reset(new TNCBufferType());
            CNCServer::WriteAppSetup(*m_SendBuff, client);
            m_SendBuff->WriteText("\n}}\nOK:END\n");
            x_SetFlag(fNoReplyOnFinish);
            m_SendPos = 0;
            return &CNCMessageHandler::x_WriteSendBuff;
        } else if (section == "storage") {
            CNCBlobStorage::WriteSetup(*this);
        } else if (section == "mirror") {
            CNCDistributionConf::WriteSetup(*this);
        } else if (section == "env") {
            CNCServer::WriteEnvInfo(*this);
            CNCBlobStorage::WriteEnvInfo(*this);
            CNCDistributionConf::WriteEnvInfo(*this);
        } else if (section == "stat") {
            CSrvRef<CNCStat> stat = CNCStat::GetStat("1min", false);
            if (stat) {
                stat->PrintState(*this);
            }
            CNCPeerControl::PrintState(*this);
#ifdef _DEBUG
        } else if (section == "syncstat") {
            CNCPeerControl::PrintSyncStat(*this);
#endif
#if __NC_TASKS_MONITOR
        } else if (section == "tasks") {
            CSrvTask::PrintState(*this);
#endif
        } else if (section == "allalerts") {
            CNCAlerts::Report(*this, true);
        } else if (section == "alerts") {
            CNCAlerts::Report(*this, false);
        } else if (section == "sync") {
            CTempString mask = params.find("port") != params.end() ? params.at("port") : CTempString(kEmptyStr);
            Flush();
            m_SendBuff.reset(new TNCBufferType());
            CNCActiveSyncControl::PrintState(*m_SendBuff, mask);
            m_SendBuff->WriteText("\n}}\nOK:END\n");
            x_SetFlag(fNoReplyOnFinish);
            m_SendPos = 0;
            return &CNCMessageHandler::x_WriteSendBuff;
        } else if (section == "db") {
            CTempString mask = params.find("port") != params.end() ? params.at("port") : CTempString(kEmptyStr);
            Flush();
            m_SendBuff.reset(new TNCBufferType());
            CNCBlobStorage::WriteDbInfo(*m_SendBuff, mask);
            m_SendBuff->WriteText("\n}}\nOK:END\n");
            x_SetFlag(fNoReplyOnFinish);
            m_SendPos = 0;
            return &CNCMessageHandler::x_WriteSendBuff;
        } else if (section == "blobs") {
            CNCBlobStorage::WriteBlobStat(*this);
        } else if (section == "blist") {
            CTempString mask = params.find("port") != params.end() ? params.at("port") : CTempString(kEmptyStr);
            Flush();
            m_SendBuff.reset(new TNCBufferType());
            CNCBlobStorage::WriteBlobList(*m_SendBuff, mask);
            m_SendBuff->WriteText("\n}}\nOK:END\n");
            x_SetFlag(fNoReplyOnFinish);
            m_SendPos = 0;
            return &CNCMessageHandler::x_WriteSendBuff;
        } else {
            WriteText(",\n\"error\": \"Unknown section name, valid names: ");
#if __NC_TASKS_MONITOR
            WriteText("task_server, netcache, storage, mirror, alerts, allalerts, env, stat, sync, tasks, blobs, db\"");
#else
            WriteText("task_server, netcache, storage, mirror, alerts, allalerts, env, stat, sync, blobs, db\"");
#endif
        }
        WriteText("\n}}");
    } else {
        CNcbiOstrstream str;
        CTaskServer::GetConfRegistry().Write(str);
        string conf = CNcbiOstrstreamToString(str);
        WriteText(conf);
    }
    x_ReportOK("\nOK:END\n");
    return &CNCMessageHandler::x_FinishCommand;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_DoCmd_AckAlert(void)
{
    LOG_CURRENT_FUNCTION
    CNCAlerts::EAlertAckResult res = CNCAlerts::eNotFound;
    TNSProtoParams& params = m_ParsedCmd.params;
    if (params.find("alert") != params.end()) {
        string alert(params["alert"]);
        if (params.find("user") != params.end()) {
            string user(params["user"]);
            res = CNCAlerts::Acknowledge(alert, user);
        }
    }
    if (res == CNCAlerts::eNotFound) {
        x_ReportError("ERR:Not found");
    } else {
        x_ReportOK("OK:END\n");
    }
    return &CNCMessageHandler::x_FinishCommand;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_DoCmd_ReConfig(void)
{
    LOG_CURRENT_FUNCTION
    if (!CNCServer::IsInitiallySynced()) {
        x_ReportError("ERR:Initial sync not finished");
        GetDiagCtx()->SetRequestStatus(eStatus_CondFailed);
        return &CNCMessageHandler::x_FinishCommand;
    }
    TNSProtoParams& params = m_ParsedCmd.params;
    string err_message("ERR:Unknown section name");
    bool result = false;
    if (params.find("section") != params.end()) {
        string section(params["section"]);
        if (section != "task_server" &&
            section != "mirror" &&
            section != "storage") {
            err_message += ": " + section;
            goto done;
        }
        CNcbiRegistry* new_reg = NULL;
        if (!CTaskServer::ReadConfiguration(new_reg))
        {
            err_message = "ERR:Failed to load registry";
            goto done;
        }
        if (section == "task_server") {
            result = CTaskServer::ReConfig(*new_reg, err_message);
        } else if (section == "mirror") {
            result = CNCDistributionConf::ReConfig(*new_reg, err_message);
            if (result) {
                CNCPeriodicSync::ReConfig();
            }
        }  else if (section == "storage") {
            result = CNCBlobStorage::ReConfig(*new_reg, err_message);
        }
        delete new_reg;
    }
done:
    if (result) {
        x_ReportOK("OK:\n");
    } else {
        x_ReportError(err_message);
    }
    return &CNCMessageHandler::x_FinishCommand;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_DoCmd_GetStat(void)
{
    LOG_CURRENT_FUNCTION
    CSrvRef<CNCStat> stat = CNCStat::GetStat(m_StatType, m_StatPrev);
    if (!stat) {
        x_ReportError("ERR:Unknown statistics type: " + m_StatType);
        GetDiagCtx()->SetRequestStatus(eStatus_BadCmd);
    }
    else {
        stat->PrintToSocket(this);
        x_ReportOK("OK:END\n");
    }
    return &CNCMessageHandler::x_FinishCommand;
}
 
CNCMessageHandler::State
CNCMessageHandler::x_DoCmd_Put(void)
{
    LOG_CURRENT_FUNCTION
    if (x_IsUserFlagSet(fNoCreate) && !(m_BlobAccess->IsBlobExists() && !m_BlobAccess->IsCurBlobExpired())) {
        return &CNCMessageHandler::x_ReportBlobNotFound;
    }
    if (!m_BlobAccess->IsBlobExists()) {
        EHTTPStatus sts = eStatus_Created;
        if (x_IsHttpMode()) {
    // if blob does not exist, verify that it is POST, not PUT
            if (!x_IsFlagSet(fCanGenerateKey)) {
                sts = eStatus_NotFound;
            }
            if (CNCBlobStorage::IsDraining()) {
                sts = eStatus_ServiceUnavailable;