NCBI C Toolkit Cross Reference

C/connect/ncbi_dispd.c


  1 /* $Id: ncbi_dispd.c,v 6.93 2009/02/04 19:29:34 kazimird Exp $
  2  * ===========================================================================
  3  *
  4  *                            PUBLIC DOMAIN NOTICE
  5  *               National Center for Biotechnology Information
  6  *
  7  *  This software/database is a "United States Government Work" under the
  8  *  terms of the United States Copyright Act.  It was written as part of
  9  *  the author's official duties as a United States Government employee and
 10  *  thus cannot be copyrighted.  This software/database is freely available
 11  *  to the public for use. The National Library of Medicine and the U.S.
 12  *  Government have not placed any restriction on its use or reproduction.
 13  *
 14  *  Although all reasonable efforts have been taken to ensure the accuracy
 15  *  and reliability of the software and data, the NLM and the U.S.
 16  *  Government do not and cannot warrant the performance or results that
 17  *  may be obtained by using this software or data. The NLM and the U.S.
 18  *  Government disclaim all warranties, express or implied, including
 19  *  warranties of performance, merchantability or fitness for any particular
 20  *  purpose.
 21  *
 22  *  Please cite the author in any work or product based on this material.
 23  *
 24  * ===========================================================================
 25  *
 26  * Author:  Anton Lavrentiev
 27  *
 28  * File Description:
 29  *   Low-level API to resolve NCBI service name to the server meta-address
 30  *   with the use of NCBI network dispatcher (DISPD).
 31  *
 32  */
 33 
 34 #include "ncbi_ansi_ext.h"
 35 #include "ncbi_comm.h"
 36 #include "ncbi_dispd.h"
 37 #include "ncbi_lb.h"
 38 #include "ncbi_priv.h"
 39 #include <connect/ncbi_connection.h>
 40 #include <connect/ncbi_http_connector.h>
 41 #include <ctype.h>
 42 #include <math.h>
 43 #include <stdio.h>
 44 #include <stdlib.h>
 45 
 46 #define NCBI_USE_ERRCODE_X   Connect_Dispd
 47 
 48 /* Lower bound of up-to-date/out-of-date ratio */
 49 #define DISPD_STALE_RATIO_OK  0.8
 50 /* Default rate increase 20% if svc runs locally */
 51 #define DISPD_LOCAL_BONUS 1.2
 52 
 53 
 54 #ifdef __cplusplus
 55 extern "C" {
 56 #endif /*__cplusplus*/
 57     static void        s_Reset      (SERV_ITER);
 58     static SSERV_Info* s_GetNextInfo(SERV_ITER, HOST_INFO*);
 59     static int/*bool*/ s_Update     (SERV_ITER, const char*, int);
 60     static void        s_Close      (SERV_ITER);
 61 
 62     static const SSERV_VTable s_op = {
 63         s_Reset, s_GetNextInfo, s_Update, 0/*Feedback*/, s_Close, "DISPD"
 64     };
 65 #ifdef __cplusplus
 66 } /* extern "C" */
 67 #endif /*__cplusplus*/
 68 
 69 
 70 struct SDISPD_Data {
 71     short/*bool*/  eof;  /* no more resolves */
 72     short/*bool*/  fail; /* no more connects */
 73     SConnNetInfo*  net_info;
 74     SLB_Candidate* cand;
 75     size_t         n_cand;
 76     size_t         a_cand;
 77     size_t         n_skip;
 78 };
 79 
 80 
 81 static int/*bool*/ s_AddServerInfo(struct SDISPD_Data* data, SSERV_Info* info)
 82 {
 83     size_t i;
 84     const char* name = SERV_NameOfInfo(info);
 85     /* First check that the new server info updates an existing one */
 86     for (i = 0; i < data->n_cand; i++) {
 87         if (strcasecmp(name, SERV_NameOfInfo(data->cand[i].info)) == 0
 88             &&  SERV_EqualInfo(info, data->cand[i].info)) {
 89             /* Replace older version */
 90             free((void*) data->cand[i].info);
 91             data->cand[i].info = info;
 92             return 1;
 93         }
 94     }
 95     /* Next, add new service to the list */
 96     if (data->n_cand == data->a_cand) {
 97         size_t n = data->a_cand + 10;
 98         SLB_Candidate* temp = (SLB_Candidate*)
 99             (data->cand
100              ? realloc(data->cand, n * sizeof(*temp))
101              : malloc (            n * sizeof(*temp)));
102         if (!temp)
103             return 0;
104         data->cand = temp;
105         data->a_cand = n;
106     }
107     data->cand[data->n_cand++].info = info;
108     return 1;
109 }
110 
111 
112 #ifdef __cplusplus
113 extern "C" {
114     static int s_ParseHeader(const char*, void*, int);
115 }
116 #endif /*__cplusplus*/
117 
118 static int/*bool*/ s_ParseHeader(const char* header,
119                                  void*       iter,
120                                  int         server_error)
121 {
122     struct SDISPD_Data* data = (struct SDISPD_Data*)((SERV_ITER) iter)->data;
123     int code = 0/*success code if any*/;
124     if (server_error) {
125         if (server_error == 400  ||  server_error == 403)
126             data->fail = 1/*true*/;
127     } else if (sscanf(header, "%*s %d", &code) < 1) {
128         data->eof = 1/*true*/;
129         return 0/*header parse error*/;
130     }
131     /* check for empty document */
132     if (!SERV_Update((SERV_ITER) iter, header, server_error)  ||  code == 204)
133         data->eof = 1/*true*/;
134     return 1/*header parsed okay*/;
135 }
136 
137 
138 #ifdef __cplusplus
139 extern "C" {
140     static int s_Adjust(SConnNetInfo*, void*, unsigned int);
141 }
142 #endif /*__cplusplus*/
143 
144 /*ARGSUSED*/
145 static int/*bool*/ s_Adjust(SConnNetInfo* net_info,
146                             void*         iter,
147                             unsigned int  n)
148 {
149     struct SDISPD_Data* data = (struct SDISPD_Data*)((SERV_ITER) iter)->data;
150     return data->fail ? 0/*no more tries*/ : 1/*may try again*/;
151 }
152 
153 
154 static void s_Resolve(SERV_ITER iter)
155 {
156     struct SDISPD_Data* data = (struct SDISPD_Data*) iter->data;
157     SConnNetInfo* net_info = data->net_info;
158     EIO_Status status = eIO_Success;
159     CONNECTOR conn = 0;
160     char* s;
161     CONN c;
162 
163     assert(!(data->eof | data->fail));
164     assert(!!net_info->stateless == !!iter->stateless);
165     /* Obtain additional header information */
166     if ((!(s = SERV_Print(iter, 0, 0))
167          ||  ConnNetInfo_OverrideUserHeader(net_info, s))
168         &&
169         ConnNetInfo_OverrideUserHeader(net_info,
170                                        iter->ok_down  &&  iter->ok_suppressed
171                                        ? "Dispatch-Mode: PROMISCUOUS\r\n"
172                                        : iter->ok_down
173                                        ? "Dispatch-Mode: OK_DOWN\r\n"
174                                        : iter->ok_suppressed
175                                        ? "Dispatch-Mode: OK_SUPPRESSED\r\n"
176                                        : "Dispatch-Mode: INFORMATION_ONLY\r\n")
177         &&
178         ConnNetInfo_OverrideUserHeader(net_info, iter->reverse_dns
179                                        ? "Client-Mode: REVERSE_DNS\r\n"
180                                        : !net_info->stateless
181                                        ? "Client-Mode: STATEFUL_CAPABLE\r\n"
182                                        : "Client-Mode: STATELESS_ONLY\r\n")) {
183         conn = HTTP_CreateConnectorEx(net_info, fHCC_SureFlush, s_ParseHeader,
184                                       s_Adjust, iter/*data*/, 0/*cleanup*/);
185     }
186     if (s) {
187         ConnNetInfo_DeleteUserHeader(net_info, s);
188         free(s);
189     }
190     if (conn  &&  (status = CONN_Create(conn, &c)) == eIO_Success) {
191         /* Send all the HTTP data, then trigger header callback */
192         CONN_Flush(c);
193         CONN_Close(c);
194     } else {
195         CORE_LOGF_X(1, eLOG_Error,
196                     ("[%s]  Unable to create auxiliary HTTP %s: %s",
197                      iter->name, conn ? "connection" : "connector",
198                      IO_StatusStr(conn ? status : eIO_Unknown)));
199         assert(0);
200     }
201 }
202 
203 
204 static int/*bool*/ s_Update(SERV_ITER iter, const char* text, int code)
205 {
206     static const char server_info[] = "Server-Info-";
207     struct SDISPD_Data* data = (struct SDISPD_Data*) iter->data;
208     int/*bool*/ failure;
209 
210     if (strncasecmp(text, server_info, sizeof(server_info) - 1) == 0
211         &&  isdigit((unsigned char) text[sizeof(server_info) - 1])) {
212         const char* name;
213         SSERV_Info* info;
214         unsigned int d1;
215         char* s;
216         int d2;
217 
218         text += sizeof(server_info) - 1;
219         if (sscanf(text, "%u: %n", &d1, &d2) < 1  ||  d1 < 1)
220             return 0/*not updated*/;
221         if (iter->ismask  ||  iter->reverse_dns) {
222             char* c;
223             if (!(s = strdup(text + d2)))
224                 return 0/*not updated*/;
225             name = s;
226             while (*name  &&  isspace((unsigned char)(*name)))
227                 name++;
228             if (!*name) {
229                 free(s);
230                 return 0/*not updated*/;
231             }
232             for (c = s + (name - s);  *c;  c++) {
233                 if (isspace((unsigned char)(*c)))
234                     break;
235             }
236             *c++ = '\0';
237             d2 += (int)(c - s);
238         } else {
239             s = 0;
240             name = "";
241         }
242         info = SERV_ReadInfoEx(text + d2, name);
243         if (s)
244             free(s);
245         if (info) {
246             if (info->time != NCBI_TIME_INFINITE)
247                 info->time += iter->time; /* expiration time now */
248             if (s_AddServerInfo(data, info))
249                 return 1/*updated*/;
250             free(info);
251         }
252     } else if (((failure = strncasecmp(text, HTTP_DISP_FAILURES,
253                                        sizeof(HTTP_DISP_FAILURES) - 1) == 0)
254                 ||  strncasecmp(text, HTTP_DISP_MESSAGES,
255                                 sizeof(HTTP_DISP_MESSAGES) - 1) == 0)  &&
256                isspace((unsigned char) text[sizeof(HTTP_DISP_FAILURES) - 1])) {
257         assert(sizeof(HTTP_DISP_FAILURES) == sizeof(HTTP_DISP_MESSAGES));
258 #if defined(_DEBUG) && !defined(NDEBUG)
259         if (data->net_info->debug_printout) {
260             text += sizeof(HTTP_DISP_FAILURES) - 1;
261             while (*text  &&  isspace((unsigned char)(*text)))
262                 text++;
263             CORE_LOGF_X(2, failure ? eLOG_Warning : eLOG_Note,
264                         ("[%s]  %s", data->net_info->service, text));
265         }
266 #endif /*_DEBUG && !NDEBUG*/
267         if (failure) {
268             if (code)
269                 data->fail = 1;
270             return 1/*updated*/;
271         }
272         /* NB: a mere message does not constitute an update */
273     }
274 
275     return 0/*not updated*/;
276 }
277 
278 
279 static int/*bool*/ s_IsUpdateNeeded(TNCBI_Time now, struct SDISPD_Data *data)
280 {
281     double status = 0.0, total = 0.0;
282 
283     if (data->n_cand) {
284         size_t i = 0;
285         while (i < data->n_cand) {
286             const SSERV_Info* info = data->cand[i].info;
287 
288             total += fabs(info->rate);
289             if (info->time < now) {
290                 if (i < --data->n_cand) {
291                     memmove(data->cand + i, data->cand + i + 1,
292                             sizeof(*data->cand)*(data->n_cand - i));
293                 }
294                 free((void*) info);
295             } else {
296                 status += fabs(info->rate);
297                 i++;
298             }
299         }
300     }
301 
302     return total == 0.0 ? 1 : status/total < DISPD_STALE_RATIO_OK;
303 }
304 
305 
306 static SLB_Candidate* s_GetCandidate(void* user_data, size_t n)
307 {
308     struct SDISPD_Data* data = (struct SDISPD_Data*) user_data;
309     return n < data->n_cand ? &data->cand[n] : 0;
310 }
311 
312 
313 static SSERV_Info* s_GetNextInfo(SERV_ITER iter, HOST_INFO* host_info)
314 {
315     struct SDISPD_Data* data = (struct SDISPD_Data*) iter->data;
316     SSERV_Info* info;
317     size_t n;
318 
319     assert(data);
320     if (!data->fail  &&  iter->n_skip < data->n_skip)
321         data->eof = 0/*false*/;
322     data->n_skip = iter->n_skip;
323 
324     if (s_IsUpdateNeeded(iter->time, data)) {
325         if (!(data->eof | data->fail))
326             s_Resolve(iter);
327         if (!data->n_cand)
328             return 0;
329     }
330 
331     for (n = 0; n < data->n_cand; n++)
332         data->cand[n].status = data->cand[n].info->rate;
333     n = LB_Select(iter, data, s_GetCandidate, DISPD_LOCAL_BONUS);
334     info = (SSERV_Info*) data->cand[n].info;
335     info->rate = data->cand[n].status;
336     if (n < --data->n_cand) {
337         memmove(data->cand + n, data->cand + n + 1,
338                 (data->n_cand - n) * sizeof(*data->cand));
339     }
340 
341     if (host_info)
342         *host_info = 0;
343     data->n_skip++;
344 
345     return info;
346 }
347 
348 
349 static void s_Reset(SERV_ITER iter)
350 {
351     struct SDISPD_Data* data = (struct SDISPD_Data*) iter->data;
352     if (data) {
353         data->eof = data->fail = 0/*false*/;
354         if (data->cand) {
355             size_t i;
356             assert(data->a_cand);
357             for (i = 0; i < data->n_cand; i++)
358                 free((void*) data->cand[i].info);
359             data->n_cand = 0;
360         }
361         data->n_skip = iter->n_skip;
362     }
363 }
364 
365 
366 static void s_Close(SERV_ITER iter)
367 {
368     struct SDISPD_Data* data = (struct SDISPD_Data*) iter->data;
369     assert(!data->n_cand); /*s_Reset() had to be called before*/
370     if (data->cand)
371         free(data->cand);
372     ConnNetInfo_Destroy(data->net_info);
373     free(data);
374     iter->data = 0;
375 }
376 
377 
378 /***********************************************************************
379  *  EXTERNAL
380  ***********************************************************************/
381 
382 /*ARGSUSED*/
383 const SSERV_VTable* SERV_DISPD_Open(SERV_ITER iter,
384                                     const SConnNetInfo* net_info,
385                                     SSERV_Info** info, HOST_INFO* u/*unused*/)
386 {
387     struct SDISPD_Data* data;
388 
389     if (!iter->ismask  &&  strpbrk(iter->name, "?*"))
390         return 0/*failed to start unallowed wildcard search*/;
391 
392     if (!(data = (struct SDISPD_Data*) calloc(1, sizeof(*data))))
393         return 0;
394     iter->data = data;
395 
396     assert(net_info); /*must be called with non-NULL*/
397     data->net_info = ConnNetInfo_Clone(net_info);
398     if (!ConnNetInfo_SetupStandardArgs(data->net_info, iter->name)) {
399         s_Close(iter);
400         return 0;
401     }
402 
403     if (g_NCBI_ConnectRandomSeed == 0) {
404         g_NCBI_ConnectRandomSeed = iter->time ^ NCBI_CONNECT_SRAND_ADDEND;
405         srand(g_NCBI_ConnectRandomSeed);
406     }
407 
408     /* Reset request method to be GET ('cause no HTTP body is ever used) */
409     data->net_info->req_method = eReqMethod_Get;
410     if (iter->stateless)
411         data->net_info->stateless = 1/*true*/;
412     if (iter->type & fSERV_Firewall)
413         data->net_info->firewall = 1/*true*/;
414     ConnNetInfo_ExtendUserHeader(data->net_info,
415                                  "User-Agent: NCBIServiceDispatcher/"
416                                  DISP_PROTOCOL_VERSION
417 #ifdef NCBI_CXX_TOOLKIT
418                                  " (C++ Toolkit)"
419 #else
420                                  " (C Toolkit)"
421 #endif /*NCBI_CXX_TOOLKIT*/
422                                  "\r\n");
423     data->n_skip = iter->n_skip;
424 
425     iter->op = &s_op; /*SERV_Update() [from HTTP callback] expects*/
426     s_Resolve(iter);
427     iter->op = 0;
428 
429     if (!data->n_cand  &&  (data->fail
430                             ||  !(data->net_info->stateless  &&
431                                   data->net_info->firewall))) {
432         s_Reset(iter);
433         s_Close(iter);
434         return 0;
435     }
436 
437     /* call GetNextInfo subsequently if info is actually needed */
438     if (info)
439         *info = 0;
440     return &s_op;
441 }
442 

source navigation ]   [ diff markup ]   [ identifier search ]   [ freetext search ]   [ file search ]  

This page was automatically generated by the LXR engine.
Visit the LXR main site for more information.