1
/******************************************************************************
3
** (c) 2012 The Johns Hopkins University Applied Physics Laboratory
4
** All rights reserved.
6
** This material may only be used, modified, or reproduced by or for the
7
** U.S. Government pursuant to the license rights granted under
8
** FAR clause 52.227-14 or DFARS clauses 252.227-7013/7014
10
** For any other permissions, please contact the Legal Office at JHU/APL.
11
******************************************************************************/
13
/*****************************************************************************
15
** File Name: ingest.h
17
** Description: This implements the data ingest thread to receive DTNMP msgs.
24
** Modification History:
25
** MM/DD/YY AUTHOR DESCRIPTION
26
** -------- ------------ ---------------------------------------------
27
** 08/31/11 V. Ramachandran Initial Implementation
28
** 01/10/13 E. Birrane Updates to lastest version of DTNMP spec.
29
** 06/27/13 E. Birrane Support persisted rules.
30
*****************************************************************************/
38
#include "shared/utils/nm_types.h"
39
#include "shared/adm/adm.h"
40
#include "shared/utils/ion_if.h"
41
#include "shared/msg/pdu.h"
43
#include "shared/msg/msg_admin.h"
44
#include "shared/msg/msg_reports.h"
45
#include "shared/msg/msg_def.h"
46
#include "shared/msg/msg_ctrl.h"
47
#include "shared/primitives/rules.h"
48
#include "shared/primitives/instr.h"
50
#include "shared/utils/utils.h"
54
extern eid_t manager_eid;
56
/******************************************************************************
58
* \par Function Name: rx_validate_mid_mc
60
* \par Determine whether a lyst contains valid, recognized MIDs.
62
* \param[in] mids The list of mids to validate.
63
* \param[in] passEmpty Whether an empty list is OK (1) or not (0)
66
* - A NULL list is always bad.
68
* \return 0 - Lyst failed to validate.
71
* Modification History:
72
* MM/DD/YY AUTHOR DESCRIPTION
73
* -------- ------------ ---------------------------------------------
74
* 01/10/13 E. Birrane Initial implementation.
75
*****************************************************************************/
77
int rx_validate_mid_mc(Lyst mids, int passEmpty)
80
mid_t *cur_mid = NULL;
83
DTNMP_DEBUG_ENTRY("rx_validate_mid_mc","(0x%#llx, %d)",
84
(unsigned long) mids, passEmpty);
86
/* Step 0 : Sanity Check. */
89
DTNMP_DEBUG_ERR("rx_validate_mid_mc", "Bad Args.", NULL);
90
DTNMP_DEBUG_EXIT("rx_validate_mid_mc","-> 0", NULL);
94
/* Step 1: Walk the list of MIDs. */
95
for (elt = lyst_first(mids); elt; elt = lyst_next(elt))
99
/* Grab the next mid...*/
100
if((cur_mid = (mid_t*) lyst_data(elt)) == NULL)
102
DTNMP_DEBUG_ERR("rx_validate_mid_mc","Found unexpected NULL mid.", NULL);
103
DTNMP_DEBUG_EXIT("rx_validate_mid_mc","-> 0", NULL);
107
char *mid_str = mid_to_string(cur_mid);
109
/* Is this a valid MID? */
110
if(mid_sanity_check(cur_mid) == 0)
112
DTNMP_DEBUG_ERR("rx_validate_mid_mc","Malformed MID: %s.", mid_str);
114
DTNMP_DEBUG_EXIT("rx_validate_mid_mc","-> 0", NULL);
118
/* Do we know this MID? */
119
if((adm_find_datadef(cur_mid) == NULL) &&
120
(def_find_by_id(gAgentVDB.reports, &(gAgentVDB.reports_mutex), cur_mid) == NULL))
122
DTNMP_DEBUG_ERR("rx_validate_mid_mc","Unknown MID %s.", mid_str);
124
DTNMP_DEBUG_EXIT("rx_validate_mid_mc","-> 0", NULL);
128
DTNMP_DEBUG_INFO("rx_validate_mid_mc","MID %s is recognized.", mid_str);
134
if((i == 0) && (passEmpty == 0))
136
DTNMP_DEBUG_ERR("rx_validate_mid_mc","Empty MID list not allowed.", NULL);
137
DTNMP_DEBUG_EXIT("rx_validate_mid_mc","-> 0", NULL);
141
DTNMP_DEBUG_EXIT("rx_validate_mid_mc","-> 1", NULL);
147
/******************************************************************************
149
* \par Function Name: rx_validate_rule
151
* \par Determines whether a production rule is correct.
153
* \param[in] rule The rule being evaluated.
155
* \return 0 - Bad rule.
158
* Modification History:
159
* MM/DD/YY AUTHOR DESCRIPTION
160
* -------- ------------ ---------------------------------------------
161
* 01/10/13 E. Birrane Initial implementation.
162
*****************************************************************************/
164
int rx_validate_rule(rule_time_prod_t *rule)
168
DTNMP_DEBUG_ENTRY("rx_validate_rule","(0x%x)", (unsigned long) rule);
170
/* Step 0: Sanity Check. */
173
DTNMP_DEBUG_ERR("rx_validate_rule","NULL rule.", NULL);
174
DTNMP_DEBUG_EXIT("rx_validate_rule","-> 0", NULL);
178
/* Is the interval correct? */
179
if(rule->desc.interval_ticks == 0)
181
DTNMP_DEBUG_ERR("rx_validate_rule","Bad interval ticks: 0.", NULL);
182
DTNMP_DEBUG_EXIT("rx_validate_rule","-> 0", NULL);
186
/* Do we understand the sender EID? */
187
if(memcmp(&(rule->desc.sender), &(manager_eid), MAX_EID_LEN) != 0)
189
DTNMP_DEBUG_ERR("rx_validate_rule","Unknown EID: %s.", rule->desc.sender.name);
190
DTNMP_DEBUG_EXIT("rx_validate_rule","-> 0", NULL);
194
/* Is each MID valid and recognized? */
195
if(rx_validate_mid_mc(rule->mids, 0) == 0)
197
DTNMP_DEBUG_ERR("rx_validate_rule","Unknown MIDs",NULL);
198
DTNMP_DEBUG_EXIT("rx_validate_rule","-> 0", NULL);
202
DTNMP_DEBUG_EXIT("rx_validate_rule","-> 1", NULL);
209
/******************************************************************************
211
* \par Function Name: rx_thread
213
* \par Receives and processes a DTNMP message.
215
* \param[in] threadId The thread identifier.
217
* \return NULL - Error
218
* !NULL - Some thread thing.
220
* Modification History:
221
* MM/DD/YY AUTHOR DESCRIPTION
222
* -------- ------------ ---------------------------------------------
223
* 01/10/13 E. Birrane Initial implementation.
224
*****************************************************************************/
226
void *rx_thread(void *threadId) {
228
DTNMP_DEBUG_ENTRY("rx_thread","(0x%x)",(unsigned long) threadId);
230
DTNMP_DEBUG_INFO("rx_thread","Receiver thread running...", NULL);
232
uint32_t num_msgs = 0;
234
uint8_t *cursor = NULL;
237
pdu_header_t *hdr = NULL;
238
pdu_acl_t *acl = NULL;
242
time_t group_timestamp = 0;
245
* g_running controls the overall execution of threads in the
250
/* Step 1: Receive a message from the Bundle Protocol Agent. */
251
buf = iif_receive(&ion_ptr, &size, &meta, NM_RECEIVE_TIMEOUT_SEC);
255
DTNMP_DEBUG_INFO("rx_thread","Received buf (%x) of size %d",
256
(unsigned long) buf, size);
258
/* Grab # messages and timestamp for this group. */
261
bytes = utils_grab_sdnv(cursor, size, &val);
266
bytes = utils_grab_sdnv(cursor, size, &val);
267
group_timestamp = val;
271
DTNMP_DEBUG_INFO("rx_thread","Group had %d msgs", num_msgs);
272
DTNMP_DEBUG_INFO("rx_thread","Group time stamp %lu", (unsigned long) group_timestamp);
274
/* For each message in the bundle. */
275
for(i = 0; i < num_msgs; i++)
277
hdr = pdu_deserialize_hdr(cursor, size, &bytes);
283
case MSG_TYPE_CTRL_PERIOD_PROD:
285
DTNMP_DEBUG_ALWAYS("NM Agent :","Received Periodic Production Message.\n", NULL);
286
rx_handle_time_prod(&meta, cursor,size,&bytes);
290
case MSG_TYPE_DEF_CUST_RPT:
292
DTNMP_DEBUG_ALWAYS("NM Agent :","Received Custom Report Definition.\n", NULL);
293
rx_handle_rpt_def(&meta, cursor,size,&bytes);
297
case MSG_TYPE_CTRL_EXEC:
299
DTNMP_DEBUG_ALWAYS("NM Agent :","Received Perform Control Message.\n", NULL);
300
rx_handle_exec(&meta, cursor,size,&bytes);
304
case MSG_TYPE_DEF_MACRO:
306
DTNMP_DEBUG_ALWAYS("NM Agent :","Received Macro Definition.\n", NULL);
307
rx_handle_macro_def(&meta, cursor,size,&bytes);
313
DTNMP_DEBUG_WARN("rx_thread","Received unknown type: %d.\n", hdr->type);
314
DTNMP_DEBUG_ALWAYS("NM Agent :","Received Unsupported message of type 0x%x.\n", hdr->id);
323
DTNMP_DEBUG_ALWAYS("rx_thread","Shutting Down Agent Receive Thread.",NULL);
324
DTNMP_DEBUG_EXIT("rx_thread","->.", NULL);
329
/******************************************************************************
331
* \par Function Name: rx_handle_rpt_def
333
* \par Process a received custom report definition message. This function
334
* accepts a portion of a serialized message group, with the
335
* understanding that the custom report definition message is at the
336
* head of the serialized data stream. This function extracts the
337
* current message, and returns the number of bytes consumed so that
338
* the called may then know where the next message in the serialized
339
* message group begins.
341
* \param[in] meta The metadata associated with the message.
342
* \param[in] cursor Pointer to the start of the serialized message.
343
* \param[in] size The size of the remaining serialized message group
344
* \param[out] bytes_used The number of bytes consumed in processing this msg.
346
* Modification History:
347
* MM/DD/YY AUTHOR DESCRIPTION
348
* -------- ------------ ---------------------------------------------
349
* 01/10/13 E. Birrane Initial implementation.
350
*****************************************************************************/
352
void rx_handle_rpt_def(pdu_metadata_t *meta, uint8_t *cursor, uint32_t size, uint32_t *bytes_used)
354
def_gen_t* rpt_def = NULL;
357
DTNMP_DEBUG_ENTRY("rx_handle_rpt_def","(0x%#llx, %d, 0x%#llx)",
358
(unsigned long)cursor, size, (unsigned long) bytes_used);
360
/* Step 0: Sanity checks. */
361
if((meta == NULL) || (cursor == NULL) || (bytes_used == NULL))
363
DTNMP_DEBUG_ERR("rx_handle_rpt_def","Bad args.",NULL);
364
DTNMP_DEBUG_EXIT("rx_handle_rpt_def","->.",NULL);
368
/* Step 1: Attempt to deserialize the message. */
369
rpt_def = def_deserialize_gen(cursor, size, &bytes);
371
/* Step 2: If the deserialization failed, complain. */
372
if((rpt_def == NULL) || (bytes == 0))
374
DTNMP_DEBUG_ERR("rx_handle_rpt_def","Can't deserialize.",NULL);
375
def_release_gen(rpt_def);
377
DTNMP_DEBUG_EXIT("rx_handle_rpt_def","->.",NULL);
381
/* Step 3: Otherwise, note how many bytes were consumed. */
384
DTNMP_DEBUG_INFO("rx_handle_rpt_def","Adding new report definition.", NULL);
387
/* Step 4: Persist this definition to our SDR. */
388
agent_db_report_persist(rpt_def);
390
/* Step 5: Persist this definition to our memory lists. */
391
// agent_vdb_reports_init(getIonsdr());
394
/* Step 6: Update instrumentation counters. */
395
gAgentInstr.num_rpt_defs++;
400
/******************************************************************************
402
* \par Function Name: rx_handle_exec
404
* \par Process a received control exec message. This function
405
* accepts a portion of a serialized message group, with the
406
* understanding that the control exec message is at the
407
* head of the serialized data stream. This function extracts the
408
* current message, and returns the number of bytes consumed so that
409
* the called may then know where the next message in the serialized
410
* message group begins.
412
* \param[in] meta The metadata associated with the message.
413
* \param[in] cursor Pointer to the start of the serialized message.
414
* \param[in] size The size of the remaining serialized message group
415
* \param[out] bytes_used The number of bytes consumed in processing this msg.
417
* Modification History:
418
* MM/DD/YY AUTHOR DESCRIPTION
419
* -------- ------------ ---------------------------------------------
420
* 01/10/13 E. Birrane Initial implementation.
421
*****************************************************************************/
423
void rx_handle_exec(pdu_metadata_t *meta, uint8_t *cursor, uint32_t size, uint32_t *bytes_used)
425
ctrl_exec_t* ctrl = NULL;
428
DTNMP_DEBUG_ENTRY("rx_handle_exec","(0x%#llx, %d, 0x%#llx)",
429
(unsigned long)cursor, size, (unsigned long) bytes_used);
432
/* Step 0: Sanity checks. */
433
if((meta == NULL) || (cursor == NULL) || (bytes_used == NULL))
435
DTNMP_DEBUG_ERR("rx_handle_exec","Bad args.",NULL);
436
DTNMP_DEBUG_EXIT("rx_handle_exec","->.",NULL);
440
/* Step 1: Attempt to deserialize the message. */
441
ctrl = ctrl_deserialize_exec(cursor, size, &bytes);
443
/* Step 2: If the deserialization failed, complain. */
444
if((ctrl == NULL) || (bytes == 0))
446
DTNMP_DEBUG_ERR("rx_handle_exec","Can't deserialize.",NULL);
447
ctrl_release_exec(ctrl);
449
DTNMP_DEBUG_EXIT("rx_handle_exec","->.",NULL);
453
/* Step 3: Otherwise, note how many bytes were consumed. */
458
* Step 4: Adjust the countdown ticks based on whether
459
* we are given a relative or absolute time.
462
if(ctrl->time <= DTNMP_RELATIVE_TIME_EPOCH)
464
/* Step 4a: If relative time, that is # seconds. */
465
ctrl->countdown_ticks = ctrl->time;
470
* Step 4b: If absolute time, # seconds if difference
471
* from now until then.
473
ctrl->countdown_ticks = (ctrl->time - getUTCTime());
476
/* Step 5: Populate dynamic parts of the control. */
477
ctrl->desc.state = CONTROL_ACTIVE;
478
strcpy(ctrl->desc.sender.name, meta->senderEid.name);
480
/* Step 6: Persist this definition to our SDR. */
481
agent_db_ctrl_persist(ctrl);
483
/* Step 7: Persist this definition to our memory lists. */
484
DTNMP_DEBUG_INFO("rx_handle_exec","Performing control.", NULL);
487
/* Step 8: Update instrumentation counters. */
488
gAgentInstr.num_ctrls++;
493
/******************************************************************************
495
* \par Function Name: rx_handle_time_prod
497
* \par Process a received time-based prod message. This function
498
* accepts a portion of a serialized message group, with the
499
* understanding that the time-based prod message is at the
500
* head of the serialized data stream. This function extracts the
501
* current message, and returns the number of bytes consumed so that
502
* the called may then know where the next message in the serialized
503
* message group begins.
505
* \param[in] meta The metadata associated with the message.
506
* \param[in] cursor Pointer to the start of the serialized message.
507
* \param[in] size The size of the remaining serialized message group
508
* \param[out] bytes_used The number of bytes consumed in processing this msg.
510
* Modification History:
511
* MM/DD/YY AUTHOR DESCRIPTION
512
* -------- ------------ ---------------------------------------------
513
* 01/10/13 E. Birrane Initial implementation.
514
*****************************************************************************/
516
void rx_handle_time_prod(pdu_metadata_t *meta, uint8_t *cursor, uint32_t size, uint32_t *bytes_used)
518
rule_time_prod_t *new_rule = NULL;
521
DTNMP_DEBUG_INFO("rx_handle_time_prod",
522
"Processing a production rule.", NULL);
524
/* Step 0: Sanity checks. */
525
if((meta == NULL) || (cursor == NULL) || (bytes_used == NULL))
527
DTNMP_DEBUG_ERR("rx_handle_time_prod","Bad args.",NULL);
528
DTNMP_DEBUG_EXIT("rx_handle_time_prod","->.",NULL);
532
/* Step 1: Attempt to deserialize the message. */
533
new_rule = ctrl_deserialize_time_prod_entry(cursor, size, &bytes);
535
/* Step 2: If the deserialization failed, complain. */
536
if((new_rule == NULL) || (bytes == 0))
538
DTNMP_DEBUG_ERR("rx_handle_time_prod","Can't deserialize.",NULL);
539
rule_release_time_prod_entry(new_rule);
541
DTNMP_DEBUG_EXIT("rx_handle_time_prod","->.",NULL);
545
/* Step 3: Otherwise, note how many bytes were consumed. */
548
/* Step 4: Populate dynamic parts of the control. */
549
/* \todo: Consider single-fire absolute-time rules. */
550
new_rule->desc.num_evals = new_rule->count;
551
new_rule->desc.interval_ticks = new_rule->period;
552
new_rule->countdown_ticks = new_rule->desc.interval_ticks;
554
strcpy(new_rule->desc.sender.name, meta->senderEid.name);
556
if(new_rule->desc.num_evals == 0)
558
new_rule->desc.num_evals = DTNMP_RULE_EXEC_ALWAYS;
561
/* Step 5: Validate the new rule. */
562
if(rx_validate_rule(new_rule) == 0)
564
DTNMP_DEBUG_ERR("rx_handle_time_prod","New rule failed validation.",NULL);
565
rule_release_time_prod_entry(new_rule);
567
DTNMP_DEBUG_EXIT("rx_handle_time_prod","->.",NULL);
571
DTNMP_DEBUG_INFO("rx_handle_time_prod",
572
"Adding new production rule.", NULL);
574
/* Step 6: Persist this definition to our SDR. */
575
agent_db_rule_persist(new_rule);
577
/* Step 7: Persist this definition to our memory lists. */
580
/* Step 8: Update instrumentation counters. */
581
gAgentInstr.num_time_rules++;
587
/******************************************************************************
589
* \par Function Name: rx_handle_macro_def
591
* \par Process a received macro def message. This function
592
* accepts a portion of a serialized message group, with the
593
* understanding that the macro def message is at the
594
* head of the serialized data stream. This function extracts the
595
* current message, and returns the number of bytes consumed so that
596
* the called may then know where the next message in the serialized
597
* message group begins.
599
* \param[in] meta The metadata associated with the message.
600
* \param[in] cursor Pointer to the start of the serialized message.
601
* \param[in] size The size of the remaining serialized message group
602
* \param[out] bytes_used The number of bytes consumed in processing this msg.
604
* Modification History:
605
* MM/DD/YY AUTHOR DESCRIPTION
606
* -------- ------------ ---------------------------------------------
607
* 01/10/13 E. Birrane Initial implementation.
608
*****************************************************************************/
609
void rx_handle_macro_def(pdu_metadata_t *meta, uint8_t *cursor, uint32_t size, uint32_t *bytes_used)
611
def_gen_t* macro_def = NULL;
614
DTNMP_DEBUG_ENTRY("rx_handle_macro_def","(0x%x, %d, 0x%x)",
615
(unsigned long)cursor, size, (unsigned long) bytes_used);
618
/* Step 0: Sanity checks. */
619
if((meta == NULL) || (cursor == NULL) || (bytes_used == NULL))
621
DTNMP_DEBUG_ERR("rx_handle_time_prod","Bad args.",NULL);
622
DTNMP_DEBUG_EXIT("rx_handle_time_prod","->.",NULL);
626
/* Step 1: Attempt to deserialize the message. */
627
macro_def = def_deserialize_gen(cursor, size, &bytes);
629
/* Step 2: If the deserialization failed, complain. */
630
if((macro_def == NULL) || (bytes == 0))
632
DTNMP_DEBUG_ERR("rx_handle_macro_def","Can;t deserialize.",NULL);
633
def_release_gen(macro_def);
635
DTNMP_DEBUG_EXIT("rx_handle_macro_def","->.",NULL);
639
/* Step 3: Otherwise, note how many bytes were consumed. */
642
DTNMP_DEBUG_INFO("rx_handle_macro_def","Adding new report definition.", NULL);
644
/* Step 4: Persist this definition to our SDR. */
645
agent_db_macro_persist(macro_def);
647
/* Step 5: Persist this definition to our memory lists. */
648
ADD_MACRO(macro_def);
650
/* Step 6: Update instrumentation counters. */
651
gAgentInstr.num_macros++;
added
removed
nm/agent/ingest.c
