ncbi_dispd.c

ncbi源码

/*
 * ===========================================================================
 * PRODUCTION $Log: ncbi_dispd.c,v $
 * PRODUCTION Revision 1000.0  2003/10/29 16:36:28  gouriano
 * PRODUCTION PRODUCTION: IMPORTED [ORIGINAL] Dev-tree R6.61
 * PRODUCTION
 * ===========================================================================
 */

/*  $Id: ncbi_dispd.c,v 1000.0 2003/10/29 16:36:28 gouriano Exp $
 * ===========================================================================
 *
 *                            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:  Anton Lavrentiev
 *
 * File Description:
 *   Low-level API to resolve NCBI service name to the server meta-address
 *   with the use of NCBI network dispatcher (DISPD).
 *
 */

#include "ncbi_ansi_ext.h"
#include "ncbi_comm.h"
#include "ncbi_dispd.h"
#include "ncbi_priv.h"
#include <connect/ncbi_connection.h>
#include <connect/ncbi_http_connector.h>
#include <ctype.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>

/* Lower bound of up-to-date/out-of-date ratio */
#define SERV_DISPD_STALE_RATIO_OK  0.8
/* Default rate increase if svc runs locally */
#define SERV_DISPD_LOCAL_SVC_BONUS 1.2


/* Dispatcher messaging support */
static int               s_MessageIssued = 0;
static FDISP_MessageHook s_MessageHook = 0;


#ifdef __cplusplus
extern "C" {
#endif
    static void        s_Reset      (SERV_ITER);
    static SSERV_Info* s_GetNextInfo(SERV_ITER, HOST_INFO*);
    static int/*bool*/ s_Update     (SERV_ITER, TNCBI_Time, const char*);
    static void        s_Close      (SERV_ITER);

    static const SSERV_VTable s_op = {
        s_Reset, s_GetNextInfo, s_Update, 0/*Penalize*/, s_Close, "DISPD"
    };
#ifdef __cplusplus
} /* extern "C" */
#endif


static int s_RandomSeed = 0;


typedef struct {
    SSERV_Info*   info;
    double        status;
} SDISPD_Node;


typedef struct {
    int/*bool*/   disp_fail;
    SConnNetInfo* net_info;
    SDISPD_Node*  s_node;
    size_t        n_node;
    size_t        n_max_node;
} SDISPD_Data;


static int/*bool*/ s_AddServerInfo(SDISPD_Data* data, SSERV_Info* info)
{
    size_t i;

    /* First check that the new server info updates an existing one */
    for (i = 0; i < data->n_node; i++) {
        if (SERV_EqualInfo(data->s_node[i].info, info)) {
            /* Replace older version */
            free(data->s_node[i].info);
            data->s_node[i].info = info;
            return 1;
        }
    }

    /* Next, add new service to the list */
    if (data->n_node == data->n_max_node) {
        size_t n = data->n_max_node + 10;
        SDISPD_Node* temp;

        if (data->s_node)
            temp = (SDISPD_Node*) realloc(data->s_node, sizeof(*temp) * n);
        else
            temp = (SDISPD_Node*) malloc(sizeof(*temp) * n);
        if (!temp)
            return 0;

        data->s_node = temp;
        data->n_max_node = n;
    }

    data->s_node[data->n_node++].info = info;
    return 1;
}


#ifdef __cplusplus
extern "C" {
    static int s_ParseHeader(const char*, void*, int);
}
#endif /* __cplusplus */

/*ARGSUSED*/
static int/*bool*/ s_ParseHeader(const char* header, void *iter,
                                 int/*ignored*/ server_error)
{
    SERV_Update((SERV_ITER) iter, header);
    return 1/*header parsed okay*/;
}


#ifdef __cplusplus
extern "C" {
    static int s_Adjust(SConnNetInfo*, void*, unsigned int);
}
#endif /* __cplusplus */

/*ARGSUSED*/
/* This callback is only for services called via direct HTTP */
static int/*bool*/ s_Adjust(SConnNetInfo* net_info,
                            void*         iter,
                            unsigned int  n)
{
    SDISPD_Data* data = (SDISPD_Data*)((SERV_ITER) iter)->data;
    return data->disp_fail ? 0/*failed*/ : 1/*try again*/;
}


static int/*bool*/ s_Resolve(SERV_ITER iter)
{
    static const char service[]  = "service";
    static const char address[]  = "address";
    static const char platform[] = "platform";
    SDISPD_Data* data = (SDISPD_Data*) iter->data;
    SConnNetInfo *net_info = data->net_info;
    CONNECTOR conn = 0;
    const char *arch;
    unsigned int ip;
    char addr[64];
    char* s;
    CONN c;

    /* Dispatcher CGI arguments (sacrifice some if they all do not fit) */
    if ((arch = CORE_GetPlatform()) != 0 && *arch)
        ConnNetInfo_PreOverrideArg(net_info, platform, arch);
    if (*net_info->client_host && !strchr(net_info->client_host, '.') &&
        (ip = SOCK_gethostbyname(net_info->client_host)) != 0 &&
        SOCK_ntoa(ip, addr, sizeof(addr)) == 0) {
        if ((s= malloc(strlen(net_info->client_host) + strlen(addr) + 3)) != 0)
            sprintf(s, "%s(%s)", net_info->client_host, addr);
        else
            s = net_info->client_host;
    } else
        s = net_info->client_host;
    if (s && *s)
        ConnNetInfo_PreOverrideArg(net_info, address, s);
    if (s != net_info->client_host)
        free(s);
    if (!ConnNetInfo_PreOverrideArg(net_info, service, iter->service)) {
        ConnNetInfo_DeleteArg(net_info, platform);
        if (!ConnNetInfo_PreOverrideArg(net_info, service, iter->service)) {
            ConnNetInfo_DeleteArg(net_info, address);
            if (!ConnNetInfo_PreOverrideArg(net_info, service, iter->service))
                return 0/*failed*/;
        }
    }
    /* Reset request method to be GET ('cause no HTTP body will follow) */
    net_info->req_method = eReqMethod_Get;
    /* Obtain additional header information */
    if ((!(s = SERV_Print(iter))
         || ConnNetInfo_OverrideUserHeader(net_info, s))                     &&
        ConnNetInfo_OverrideUserHeader(net_info, net_info->stateless
                                       ?"Client-Mode: STATELESS_ONLY\r\n"
                                       :"Client-Mode: STATEFUL_CAPABLE\r\n") &&
        ConnNetInfo_OverrideUserHeader(net_info,
                                       "Dispatch-Mode: INFORMATION_ONLY\r\n")){
        ConnNetInfo_OverrideUserHeader
            (net_info, "User-Agent: NCBIServiceDispatcher/"
             DISP_PROTOCOL_VERSION
#ifdef NCBI_CXX_TOOLKIT
             " (C++ Toolkit)"
#else
             " (C Toolkit)"
#endif
             "\r\n");
        data->disp_fail = 0;
        /* All the rest in the net_info structure is fine with us */
        conn = HTTP_CreateConnectorEx(net_info, fHCC_SureFlush, s_ParseHeader,
                                      s_Adjust, iter/*data*/, 0/*cleanup*/);
    }
    if (s) {
        ConnNetInfo_DeleteUserHeader(net_info, s);
        free(s);
    }
    if (!conn || CONN_Create(conn, &c) != eIO_Success) {
        CORE_LOGF(eLOG_Error, ("[DISPATCHER]  Unable to create aux. %s",
                               conn ? "connection" : "connector"));
        assert(0);
        return 0/*failed*/;
    }
    /* This will also send all the HTTP data, and trigger header callback */
    CONN_Flush(c);
    CONN_Close(c);
    return ((SDISPD_Data*) iter->data)->n_node != 0;
}


static int/*bool*/ s_Update(SERV_ITER iter, TNCBI_Time now, const char* text)
{
    static const char server_info[] = "Server-Info-";
    SDISPD_Data* data = (SDISPD_Data*) iter->data;
    size_t len = strlen(text);

    if (len >= sizeof(server_info) &&
        strncasecmp(text, server_info, sizeof(server_info) - 1) == 0) {
        const char* p = text + sizeof(server_info) - 1;
        SSERV_Info* info;
        unsigned int d1;
        int d2;

        if (sscanf(p, "%u: %n", &d1, &d2) < 1)
            return 0/*not updated*/;
        if ((info = SERV_ReadInfo(p + d2)) != 0) {
            assert(info->rate != 0.0);
            info->time += now; /* expiration time now */
            if (s_AddServerInfo(data, info))
                return 1/*updated*/;
            free(info);
        }
    } else if (len >= sizeof(HTTP_DISP_FAILURES) &&
               strncasecmp(text, HTTP_DISP_FAILURES,
                           sizeof(HTTP_DISP_FAILURES) - 1) == 0) {
#if defined(_DEBUG) && !defined(NDEBUG)
        const char* p = text + sizeof(HTTP_DISP_FAILURES) - 1;
        while (*p && isspace((unsigned char)(*p)))
            p++;
        if (data->net_info->debug_printout)
            CORE_LOGF(eLOG_Warning, ("[DISPATCHER]  %s", p));
#endif
        data->disp_fail = 1;
        return 1/*updated*/;
    } else if (len >= sizeof(HTTP_DISP_MESSAGE) &&
               strncasecmp(text, HTTP_DISP_MESSAGE,
                           sizeof(HTTP_DISP_MESSAGE) - 1) == 0) {
        const char* p = text + sizeof(HTTP_DISP_MESSAGE) - 1;
        while (*p && isspace((unsigned char)(*p)))
            p++;
        if (s_MessageHook) {
            if (s_MessageIssued <= 0) {
                s_MessageIssued = 1;
                s_MessageHook(p);
            }
        } else {
            s_MessageIssued = -1;
            CORE_LOGF(eLOG_Warning, ("[DISPATCHER]  %s", p));
        }
    }

    return 0/*not updated*/;
}


static int/*bool*/ s_IsUpdateNeeded(SDISPD_Data *data)
{
    double status = 0.0, total = 0.0;

    if (data->n_node) {
        TNCBI_Time t = (TNCBI_Time) time(0);
        size_t i = 0;
        while (i < data->n_node) {
            SSERV_Info* info = data->s_node[i].info;

            total += info->rate;
            if (info->time < t) {
                if (i < --data->n_node)
                    memmove(data->s_node + i, data->s_node + i + 1,
                            (data->n_node - i)*sizeof(*data->s_node));
                free(info);
            } else {
                status += info->rate;
                i++;
            }
        }
    }

    return total == 0.0 ? 1 : (status/total < SERV_DISPD_STALE_RATIO_OK);
}


static SSERV_Info* s_GetNextInfo(SERV_ITER iter, HOST_INFO* host_info)
{
    double total = 0.0, point = 0.0, access = 0.0, p = 0.0, status;
    SDISPD_Data* data = (SDISPD_Data*) iter->data;
    SSERV_Info* info;
    size_t i;

    if (!data)
        return 0;