| 3585 | } |
| 3586 | |
| 3587 | static int process_user_message(netinfo_type *netinfo_ptr, |
| 3588 | host_node_type *host_node_ptr) |
| 3589 | { |
| 3590 | int usertype, seqnum, datalen, needack; |
| 3591 | ack_state_type *ack_state = NULL; |
| 3592 | void *data; |
| 3593 | |
| 3594 | /* deliver nothing for fake netinfo */ |
| 3595 | if (netinfo_ptr->fake || netinfo_ptr->exiting) |
| 3596 | return 0; |
| 3597 | |
| 3598 | |
| 3599 | int malloced = 0; |
| 3600 | |
| 3601 | int rc = read_user_data(host_node_ptr, &usertype, &seqnum, &needack, |
| 3602 | &datalen, &data, &malloced); |
| 3603 | |
| 3604 | #if 0 |
| 3605 | logmsg(LOGMSG_DEBUG, "process_user_message from %s, ut=%d\n", |
| 3606 | host_node_ptr->host, usertype); |
| 3607 | #endif |
| 3608 | |
| 3609 | if (rc != 0) |
| 3610 | return -1; /* not sure ... exit the reader thread??? */ |
| 3611 | |
| 3612 | if (usertype > USER_TYPE_MIN && usertype < USER_TYPE_MAX && |
| 3613 | netinfo_ptr->userfuncs[usertype].func != NULL) { |
| 3614 | if (needack) { |
| 3615 | ack_state = HOST_MALLOC(host_node_ptr, sizeof(ack_state_type)); |
| 3616 | ack_state->seqnum = seqnum; |
| 3617 | ack_state->needack = needack; |
| 3618 | ack_state->fromhost = host_node_ptr->host; |
| 3619 | ack_state->netinfo = netinfo_ptr; |
| 3620 | } else { |
| 3621 | ack_state = NULL; |
| 3622 | } |
| 3623 | |
| 3624 | Pthread_mutex_lock(&(host_node_ptr->timestamp_lock)); |
| 3625 | host_node_ptr->running_user_func = 1; |
| 3626 | Pthread_mutex_unlock(&(host_node_ptr->timestamp_lock)); |
| 3627 | |
| 3628 | int64_t start_us = comdb2_time_epochus(); |
| 3629 | /* run the user's function */ |
| 3630 | netinfo_ptr->userfuncs[usertype].func( |
| 3631 | ack_state, netinfo_ptr->usrptr, host_node_ptr->host, host_node_ptr->host_interned, |
| 3632 | usertype, data, datalen, 1); |
| 3633 | netinfo_ptr->userfuncs[usertype].count++; |
| 3634 | netinfo_ptr->userfuncs[usertype].totus += |
| 3635 | (comdb2_time_epochus() - start_us); |
| 3636 | |
| 3637 | /* update timestamp before checking it */ |
| 3638 | Pthread_mutex_lock(&(host_node_ptr->timestamp_lock)); |
| 3639 | host_node_ptr->timestamp = time(NULL); |
| 3640 | host_node_ptr->running_user_func = 0; |
| 3641 | Pthread_mutex_unlock(&(host_node_ptr->timestamp_lock)); |
| 3642 | } else { |
| 3643 | static int lastpr = 0, count = 0; |
| 3644 | int now; |
no test coverage detected