NCBI C++ ToolKit
ipsgs_processor.cpp
Go to the documentation of this file.

Go to the SVN repository for this file.

1 /* $Id: ipsgs_processor.cpp 101564 2024-01-04 19:43:03Z satskyse $
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  * Authors: Sergey Satskiy
27  *
28  * File Description: PSG processor interface
29  *
30  */
31 
32 #include <ncbi_pch.hpp>
33 
34 #include "ipsgs_processor.hpp"
35 #include "pubseq_gateway.hpp"
36 #include "insdc_utils.hpp"
37 #include "psgs_seq_id_utils.hpp"
38 
39 extern bool g_AllowProcessorTiming;
40 
41 
42 string
44 {
45  switch (status) {
47  return "ePSGS_InProgress";
49  return "ePSGS_Done";
51  return "ePSGS_NotFound";
53  return "ePSGS_Error";
55  return "ePSGS_Canceled";
57  return "ePSGS_Timeout";
59  return "ePSGS_Unauthorized";
60  default:
61  break;
62  }
63  return "unknown (" + to_string(status) + ")";
64 }
65 
66 
67 string
69 {
70  switch (status) {
72  // Note: must not really be called while a processor in this status
73  return "inprogress";
75  return "done";
77  return "not_found";
79  return "error";
81  return "canceled";
83  return "timeout";
85  return "unauthorized";
86  default:
87  break;
88  }
89  return "unknown (" + to_string(status) + ")";
90 }
91 
92 
95 {
98  // Memorize the first time only
99  m_SignalStartTimestamp = psg_clock_t::now();
101  }
102  }
103 
105 }
106 
107 
109 {
112  // Memorize the first time only
113  m_SignalFinishTimestamp = psg_clock_t::now();
115  }
116  }
117 
118  if (!m_FinishSignalled) {
121  m_FinishSignalled = true;
122  }
123 }
124 
125 
127  void * user_data)
128 {
129  auto * app = CPubseqGatewayApp::GetInstance();
130  uv_thread_t uv_thread_id = GetUVThreadId();
131 
132  if (uv_thread_id == 0) {
133  // The processor has not started yet. There is no uv loop (and thread)
134  // to bind to.
135  string msg = "Processor '" + GetName() + "' "
136  "tries to schedule a postponed callback before "
137  "a thread was assigned to the processor (request id: " +
138  to_string(m_Request->GetRequestId()) + ").";
139  PSG_ERROR(msg);
140  NCBI_THROW(CPubseqGatewayException, eLogic, msg);
141  }
142 
143  try {
144  app->GetUvLoopBinder(uv_thread_id).PostponeInvoke(cb, user_data,
145  m_Request->GetRequestId());
146  } catch (const exception & exc) {
147  PSG_ERROR("Error scheduling a postponed callback by the processor '" +
148  GetName() + "' (while serving request id: " +
149  to_string(m_Request->GetRequestId()) + "): " + exc.what());
150  throw;
151  } catch (...) {
152  PSG_ERROR("Unknown error scheduling a postponed callback by the processor '" +
153  GetName() + "' (while serving request id: " +
154  to_string(m_Request->GetRequestId()) + ")");
155  throw;
156  }
157 }
158 
159 
162  uint64_t timeout_millisec,
163  void * user_data,
167 {
168  auto * app = CPubseqGatewayApp::GetInstance();
169  uv_thread_t uv_thread_id = GetUVThreadId();
170 
171  if (uv_thread_id == 0) {
172  // The processor has not started yet. There is no uv loop (and thread)
173  // to bind to.
174  string msg = "Processor '" + GetName() + "' "
175  "tries to schedule a socket callback before "
176  "a thread was assigned to the processor (request id: " +
177  to_string(m_Request->GetRequestId()) + ").";
178  PSG_ERROR(msg);
179  NCBI_THROW(CPubseqGatewayException, eLogic, msg);
180  }
181 
182  try {
183  app->GetUvLoopBinder(uv_thread_id).SetSocketCallback(
184  fd, event, timeout_millisec, user_data,
185  event_cb, timeout_cb, error_cb,
186  m_Request->GetRequestId());
187  } catch (const exception & exc) {
188  PSG_ERROR("Error scheduling a socket callback by the processor '" +
189  GetName() + "' (while serving request id: " +
190  to_string(m_Request->GetRequestId()) + "): " + exc.what());
191  throw;
192  } catch (...) {
193  PSG_ERROR("Unknown error scheduling a socket callback by the processor '" +
194  GetName() + "' (while serving request id: " +
195  to_string(m_Request->GetRequestId()) + ")");
196  throw;
197  }
198 }
199 
200 
202  const CSeq_id& parsed_seq_id,
203  int request_seq_id_type,
204  int16_t& eff_seq_id_type,
205  bool need_trace)
206 {
207  auto parsed_seq_id_type = parsed_seq_id.Which();
208  bool parsed_seq_id_type_found = (parsed_seq_id_type !=
210 
211  if (!parsed_seq_id_type_found && request_seq_id_type < 0) {
212  eff_seq_id_type = -1;
213  return true;
214  }
215 
216  if (!parsed_seq_id_type_found) {
217  eff_seq_id_type = request_seq_id_type;
218  return true;
219  }
220 
221  if (request_seq_id_type < 0) {
222  eff_seq_id_type = parsed_seq_id_type;
223  return true;
224  }
225 
226  // Both found
227  if (parsed_seq_id_type == request_seq_id_type) {
228  eff_seq_id_type = request_seq_id_type;
229  return true;
230  }
231 
232  // The parsed and url explicit seq_id_type do not match
233  if (IsINSDCSeqIdType(parsed_seq_id_type) &&
234  IsINSDCSeqIdType(request_seq_id_type)) {
235  if (need_trace) {
236  m_Reply->SendTrace(
237  "Seq id type mismatch. Parsed CSeq_id reports seq_id_type as " +
238  to_string(parsed_seq_id_type) + " while the URL reports " +
239  to_string(request_seq_id_type) + ". They both belong to INSDC types so "
240  "CSeq_id provided type " + to_string(parsed_seq_id_type) +
241  " is taken as an effective one",
242  m_Request->GetStartTimestamp());
243  }
244  eff_seq_id_type = parsed_seq_id_type;
245  return true;
246  }
247 
248  return false;
249 }
250 
251 
253  CSeq_id& seq_id,
254  const string& request_seq_id,
255  int request_seq_id_type,
256  string* err_msg)
257 {
258  bool need_trace = m_Request->NeedTrace();
259  string stripped_seq_id = StripTrailingVerticalBars(request_seq_id);
260 
261  try {
262  seq_id.Set(request_seq_id);
263  if (need_trace)
264  m_Reply->SendTrace("Parsing CSeq_id('" + stripped_seq_id +
265  "') succeeded", m_Request->GetStartTimestamp());
266 
267  if (request_seq_id_type <= 0) {
268  if (need_trace)
269  m_Reply->SendTrace("Parsing CSeq_id finished OK (#1)",
270  m_Request->GetStartTimestamp());
271  return ePSGS_ParsedOK;
272  }
273 
274  // Check the parsed type with the given
275  int16_t eff_seq_id_type;
276  if (GetEffectiveSeqIdType(seq_id, request_seq_id_type, eff_seq_id_type, false)) {
277  if (need_trace)
278  m_Reply->SendTrace("Parsing CSeq_id finished OK (#2)",
279  m_Request->GetStartTimestamp());
280  return ePSGS_ParsedOK;
281  }
282 
283  // seq_id_type from URL and from CSeq_id differ
284  CSeq_id_Base::E_Choice seq_id_type = seq_id.Which();
285 
286  if (need_trace)
287  m_Reply->SendTrace("CSeq_id provided type " + to_string(seq_id_type) +
288  " and URL provided seq_id_type " +
289  to_string(request_seq_id_type) + " mismatch",
290  m_Request->GetStartTimestamp());
291 
292  if (IsINSDCSeqIdType(request_seq_id_type) &&
293  IsINSDCSeqIdType(seq_id_type)) {
294  // Both seq_id_types belong to INSDC
295  if (need_trace) {
296  m_Reply->SendTrace("Both types belong to INSDC types.\n"
297  "Parsing CSeq_id finished OK (#3)",
298  m_Request->GetStartTimestamp());
299  }
300  return ePSGS_ParsedOK;
301  }
302 
303  // Type mismatch: form the error message in case of resolution problems
304  if (err_msg) {
305  *err_msg = "Seq_id '" + request_seq_id +
306  "' possible type mismatch: the URL provides " +
307  to_string(request_seq_id_type) +
308  " while the CSeq_Id detects it as " +
309  to_string(static_cast<int>(seq_id_type));
310  }
311  } catch (...) {
312  if (need_trace)
313  m_Reply->SendTrace("Parsing CSeq_id('" + stripped_seq_id +
314  "') failed (exception)",
315  m_Request->GetStartTimestamp());
316  }
317 
318  // Second variation of Set()
319  if (request_seq_id_type > 0) {
320  try {
322  (CSeq_id_Base::E_Choice)(request_seq_id_type),
323  stripped_seq_id);
324  if (need_trace) {
325  m_Reply->SendTrace("Parsing CSeq_id(eFasta_AsTypeAndContent, " +
326  to_string(request_seq_id_type) +
327  ", '" + stripped_seq_id + "') succeeded.\n"
328  "Parsing CSeq_id finished OK (#4)",
329  m_Request->GetStartTimestamp());
330  }
331  return ePSGS_ParsedOK;
332  } catch (...) {
333  if (need_trace)
334  m_Reply->SendTrace("Parsing CSeq_id(eFasta_AsTypeAndContent, " +
335  to_string(request_seq_id_type) +
336  ", '" + stripped_seq_id + "') failed (exception)",
337  m_Request->GetStartTimestamp());
338  }
339  }
340 
341  if (need_trace) {
342  m_Reply->SendTrace("Parsing CSeq_id finished FAILED",
343  m_Request->GetStartTimestamp());
344  }
345 
346  return ePSGS_ParseFailed;
347 }
348 
349 
351 {
352  // Memorize the moment when the processor has started
353  // if (g_AllowProcessorTiming) {
354  // Note: it was conditional depending on g_AllowProcessorTiming
355  // Now it is unconditional because it is also used to collect per processor
356  // performance timing.
357 
358  m_ProcessInvokeTimestamp = psg_clock_t::now();
360 }
361 
362 
364  const CSeq_id& parsed_seq_id,
365  int request_seq_id_type)
366 {
367  auto parsed_seq_id_type = parsed_seq_id.Which();
368  bool parsed_seq_id_type_found = (parsed_seq_id_type !=
370 
371  if (!parsed_seq_id_type_found && request_seq_id_type < 0) {
372  return true;
373  }
374 
375  if (!parsed_seq_id_type_found) {
376  return true;
377  }
378 
379  if (request_seq_id_type < 0) {
380  return true;
381  }
382 
383  // Both found
384  if (parsed_seq_id_type == request_seq_id_type) {
385  return true;
386  }
387 
388  // The parsed and url explicit seq_id_type do not match
389  if (IsINSDCSeqIdType(parsed_seq_id_type) &&
390  IsINSDCSeqIdType(request_seq_id_type)) {
391  return true;
392  }
393 
394  return false;
395 }
396 
397 
function< EPSGS_PollContinue(void *user_data)> TEventCB
function< EPSGS_PollContinue(void *user_data)> TTimeoutCB
function< EPSGS_PollContinue(const string &message, void *user_data)> TErrorCB
function< void(void *user_data)> TProcessorCB
IPSGS_Processor::EPSGS_StartProcessing SignalStartProcessing(IPSGS_Processor *processor)
void SignalFinishProcessing(IPSGS_Processor *processor, CPSGS_Dispatcher::EPSGS_SignalSource signal_source)
static CPubseqGatewayApp * GetInstance(void)
virtual string GetName(void) const =0
Tells the processor name (used in logging and tracing)
static string StatusToString(EPSGS_Status status)
Converts the processor status to a string for tracing and logging purposes.
psg_time_point_t m_ProcessInvokeTimestamp
void OnBeforeProcess(void)
Called just before the virtual Process() method is called.
psg_time_point_t m_SignalStartTimestamp
void SetSocketCallback(int fd, CPSGS_SocketIOCallback::EPSGS_Event event, uint64_t timeout_millisec, void *user_data, CPSGS_SocketIOCallback::TEventCB event_cb, CPSGS_SocketIOCallback::TTimeoutCB timeout_cb, CPSGS_SocketIOCallback::TErrorCB error_cb)
The provided callbacks will be called from the libuv loop assigned to the processor when the correspo...
uv_thread_t GetUVThreadId(void) const
Provides the libuv thread id which runs the processor.
shared_ptr< CPSGS_Reply > m_Reply
EPSGS_StartProcessing
Tells wether to continue or not after a processor called SignalStartProcessing() method.
bool m_SignalStartTimestampInitialized
bool GetEffectiveSeqIdType(const objects::CSeq_id &parsed_seq_id, int request_seq_id_type, int16_t &eff_seq_id_type, bool need_trace)
psg_time_point_t m_SignalFinishTimestamp
EPSGS_Status
The GetStatus() method returns a processor current status.
void PostponeInvoke(CPSGS_UvLoopBinder::TProcessorCB cb, void *user_data)
The provided callback will be called from the libuv loop assigned to the processor.
bool m_ProcessInvokeTimestampInitialized
void SignalFinishProcessing(void)
A processor should call this method when it decides that there is nothing else to be done.
EPSGS_StartProcessing SignalStartProcessing(void)
A processor should call the method when it decides that it successfully started processing the reques...
shared_ptr< CPSGS_Request > m_Request
bool m_SignalFinishTimestampInitialized
EPSGS_SeqIdParsingResult ParseInputSeqId(objects::CSeq_id &seq_id, const string &request_seq_id, int request_seq_id_type, string *err_msg=nullptr)
Parse seq-id from a string and type representation.
static string StatusToProgressMessage(EPSGS_Status status)
Converts the processor status to a string for protocol message.
#define NCBI_THROW(exception_class, err_code, message)
Generic macro to throw an exception, given the exception class, error code and message string.
Definition: ncbiexpt.hpp:704
CSeq_id & Set(const CTempString &the_id, TParseFlags flags=fParse_AnyRaw)
Reassign based on flat specifications; arguments interpreted as with constructors.
Definition: Seq_id.cpp:2456
@ eFasta_AsTypeAndContent
Definition: Seq_id.hpp:117
E_Choice Which(void) const
Which variant is currently selected.
Definition: Seq_id_.hpp:746
E_Choice
Choice variants.
Definition: Seq_id_.hpp:93
@ e_not_set
No variant selected.
Definition: Seq_id_.hpp:94
bool IsINSDCSeqIdType(CBioseqInfoRecord::TSeqIdType seq_id_type)
Definition: insdc_utils.cpp:44
bool AreSeqIdTypesMatched(const CSeq_id &parsed_seq_id, int request_seq_id_type)
bool g_AllowProcessorTiming
string StripTrailingVerticalBars(const string &seq_id)
#define PSG_ERROR(message)
EPSGS_SeqIdParsingResult
@ ePSGS_ParsedOK
@ ePSGS_ParseFailed
signed short int16_t
Definition: stdint.h:122
unsigned __int64 uint64_t
Definition: stdint.h:136
Modified on Fri Mar 01 10:06:15 2024 by modify_doxy.py rev. 669887