<commit_before>#include <ctype.h>
#include <errno.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include "b2bua_socket.h"
#include "ss_lthread.h"
#include "ss_network.h"
#include "ss_util.h"

struct lthread
{
    struct lthread_args args;
    pthread_t rx_thread[10];
    struct lthread *next;
    int status;
};

#define LLTHREAD_RUNS 0
#define LLTHREAD_DEAD -1

static int lthread_sock_prepare(struct lthread_args *);
static void lthread_rx(struct lthread_args *args);
static void lthread_tx(struct lthread_args *args);

static ssize_t
recvfromto(int s, void *buf, size_t len, struct sockaddr *from,
  socklen_t *fromlen, struct sockaddr *to, socklen_t *tolen)
{
    char cbuf[CMSG_SPACE(sizeof(struct sockaddr_storage))];
    struct cmsghdr *cmsg;
    struct msghdr msg;
    struct iovec iov;
    ssize_t rval;

    memset(&msg, '\0', sizeof(msg));
    iov.iov_base = buf;
    iov.iov_len = len;
    msg.msg_name = from;
    msg.msg_namelen = *fromlen;
    msg.msg_iov = &iov;
    msg.msg_iovlen = 1;
    msg.msg_control = cbuf;
    msg.msg_controllen = CMSG_LEN(sizeof(struct sockaddr_storage));

    rval = recvmsg(s, &msg, 0);
    if (rval < 0)
        return (rval);

    *tolen = 0; 
    for (cmsg = CMSG_FIRSTHDR(&msg); cmsg != NULL;
      cmsg = CMSG_NXTHDR(&msg, cmsg)) {
        if (cmsg->cmsg_level == IPPROTO_IP &&
          cmsg->cmsg_type == IP_RECVDSTADDR) {
            memcpy(&satosin(to)->sin_addr, CMSG_DATA(cmsg),
              sizeof(struct in_addr));
            to->sa_family = AF_INET;
            *tolen = sizeof(struct sockaddr_in);
            break;
        }
    }
    *fromlen = msg.msg_namelen;
    return (rval);
}

void
lthread_mgr_run(struct lthread_args *args)
{
    struct lthread *lthread_head, *lthread;
    struct wi *wi;
    int i;

    lthread = malloc(sizeof(*lthread));
    memset(lthread, '\0', sizeof(*lthread));
    if (lthread == NULL) {
        fprintf(stderr, "lthread_mgr_run: out of mem\n");
        exit(1);
    }
    lthread->args.listen_addr = args->listen_addr;
    lthread->args.listen_port = args->listen_port;
    lthread->args.wildcard = 1;
    if (lthread_sock_prepare(&lthread->args) != 0) {
        fprintf(stderr, "lthread_sock_prepare(%s:%d) = -1\n", args->listen_addr, args->listen_port);
        exit(1);
    }
    pthread_cond_init(&lthread->args.outpacket_queue.cond, NULL);
    pthread_mutex_init(&lthread->args.outpacket_queue.mutex, NULL);
    lthread->args.outpacket_queue.name = strdup("B2B->NET (wildcard)");
    lthread->args.bslots = args->bslots;

    i = pthread_create(&lthread->rx_thread[0], NULL, (void *(*)(void *))&lthread_rx, &lthread->args);
    printf("%d\n", i);
    lthread_head = lthread;
    for (;;) {
        wi = queue_get_item(&args->outpacket_queue, 0);
        for (lthread = lthread_head; lthread != NULL; lthread = lthread->next) {
            if (lthread->args.wildcard != 0)
                continue;
            if (lthread->args.listen_port != OUTP(wi).local_port ||
              strcmp(lthread->args.listen_addr, OUTP(wi).local_addr) != 0)
                continue;
            break;
        }
        if (lthread == NULL) {
            lthread = malloc(sizeof(*lthread));
            if (lthread == NULL) {
                fprintf(stderr, "lthread_mgr_run: out of mem\n");
                wi_free(wi);
                continue;
            }
            memset(lthread, '\0', sizeof(*lthread));
            lthread->args.listen_addr = strdup(OUTP(wi).local_addr);
            lthread->args.listen_port = OUTP(wi).local_port;
            if (lthread_sock_prepare(&lthread->args) != 0) {
                fprintf(stderr, "lthread_sock_prepare(%s:%d) = -1\n", lthread->args.listen_addr, lthread->args.listen_port);
                wi_free(wi);
                free(lthread->args.listen_addr);
                free(lthread);
                continue;
            }
            pthread_cond_init(&lthread->args.outpacket_queue.cond, NULL);
            pthread_mutex_init(&lthread->args.outpacket_queue.mutex, NULL);
            asprintf(&lthread->args.outpacket_queue.name, "B2B->NET (%s:%d)", lthread->args.listen_addr,
              lthread->args.listen_port);
            lthread->args.bslots = args->bslots;

            for (i = 0; i < 10; i++) {
                pthread_create(&lthread->rx_thread[i], NULL, (void *(*)(void *))&lthread_rx, &lthread->args);
            }
            lthread->next = lthread_head;
            lthread_head = lthread;
        }
        queue_put_item(wi, &lthread->args.outpacket_queue);
    }
}

extern char *strcasestr (__const char *__haystack, __const char *__needle);

static int
extract_call_id(const char *buf, str *call_id)
{
    const char *cp, *cp1;
    int len;

    for (cp = buf;;) {
        cp1 = strcasestr(cp, "call-id:");
        if (cp1 == NULL) {
            cp = strcasestr(cp, "i:");
            if (cp == NULL)
                return -1;
            len = 2;
        } else {
            cp = cp1;
            len = 8;
        }
        if (cp > buf && cp[-1] != '\n' && cp[-1] != '\r') {
            cp += len;
            continue;
        }
        call_id->s = (char *)cp + len;
        while (call_id->s[0] != '\0' && call_id->s[0] != '\r' &&
          call_id->s[0] != '\n' && isspace(call_id->s[0]))
            call_id->s++;
        if (call_id->s[0] == '\0' || call_id->s[0] == '\r' || call_id->s[0] == '\n')
            return -1;
        call_id->len = 0;
        while (call_id->s[call_id->len] != '\0' && call_id->s[call_id->len] != '\r' &&
          call_id->s[call_id->len] != '\n' && !isspace(call_id->s[call_id->len]))
            call_id->len++;
        if (call_id->len == 0)
            return -1;
        return 0;
    }
}

static int
lthread_sock_prepare(struct lthread_args *args)
{
    struct sockaddr_storage ia;
    char listen_port[10];
    int n, on;

    sprintf(listen_port, "%d", args->listen_port);
    n = resolve(sstosa(&ia), AF_INET, args->listen_addr, listen_port, AI_PASSIVE);
    if (n != 0)
        return -1;
    args->sock = socket(AF_INET, SOCK_DGRAM, 0);
    if (args->sock < 0)
        return -1;
    setsockopt(args->sock, SOL_SOCKET, SO_REUSEADDR, &args->sock, sizeof args->sock);
    if (args->wildcard != 0) {
        on = 1;
        setsockopt(args->sock, IPPROTO_IP, IP_RECVDSTADDR, &on, sizeof(on));
    }
    n = bind(args->sock, sstosa(&ia), SS_LEN(&ia));
    if (n != 0) {
        printf("lthread_sock_prepare:bind(%d)\n", n);
        close(args->sock);
        return -1;
    }
    return 0;
}

static void
lthread_rx(struct lthread_args *args)
{
    int n;
    socklen_t ralen, lalen;
    size_t rsize;
    struct sockaddr_storage ia, la;
    struct wi *wi;
    pthread_t tx_thread;
    str call_id;
    struct b2bua_slot *bslot;

    if (args->wildcard == 0)<commit_msg>Use IP_PKTINFO option to get datagram destination address reliably.<commit_after>#include <ctype.h>
#include <errno.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include "b2bua_socket.h"
#include "ss_lthread.h"
#include "ss_network.h"
#include "ss_util.h"

struct lthread
{
    struct lthread_args args;
    pthread_t rx_thread[10];
    struct lthread *next;
    int status;
};

#define LLTHREAD_RUNS 0
#define LLTHREAD_DEAD -1

static int lthread_sock_prepare(struct lthread_args *);
static void lthread_rx(struct lthread_args *args);
static void lthread_tx(struct lthread_args *args);

static ssize_t
recvfromto(int s, void *buf, size_t len, struct sockaddr *from,
  socklen_t *fromlen, struct sockaddr *to, socklen_t *tolen)
{
#if defined(__FreeBSD__)
    char cbuf[CMSG_SPACE(sizeof(struct sockaddr_storage))];
#else
    char cbuf[CMSG_SPACE(sizeof(struct in_pktinfo))];
    struct in_pktinfo *pktinfo;
#endif
    struct cmsghdr *cmsg;
    struct msghdr msg;
    struct iovec iov;
    ssize_t rval;

    memset(&msg, '\0', sizeof(msg));
    iov.iov_base = buf;
    iov.iov_len = len;
    msg.msg_name = from;
    msg.msg_namelen = *fromlen;
    msg.msg_iov = &iov;
    msg.msg_iovlen = 1;
    msg.msg_control = cbuf;
    msg.msg_controllen = CMSG_LEN(sizeof(struct sockaddr_storage));

    rval = recvmsg(s, &msg, 0);
    if (rval < 0)
        return (rval);

    *tolen = 0; 
    for (cmsg = CMSG_FIRSTHDR(&msg); cmsg != NULL;
      cmsg = CMSG_NXTHDR(&msg, cmsg)) {
#if defined(__FreeBSD__)
        if (cmsg->cmsg_level == IPPROTO_IP &&
          cmsg->cmsg_type == IP_RECVDSTADDR) {
            memcpy(&satosin(to)->sin_addr, CMSG_DATA(cmsg),
              sizeof(struct in_addr));
            to->sa_family = AF_INET;
            *tolen = sizeof(struct sockaddr_in);
            break;
        }
#else
        if (cmsg->cmsg_level == SOL_IP &&
          cmsg->cmsg_type == IP_PKTINFO) {
            pktinfo = (struct in_pktinfo *)CMSG_DATA(cmsg);
            memcpy(&satosin(to)->sin_addr, &pktinfo->ipi_addr,
              sizeof(struct in_addr));
            to->sa_family = AF_INET;
            *tolen = sizeof(struct sockaddr_in);
            break;
        }
#endif
    }
    *fromlen = msg.msg_namelen;
    return (rval);
}

void
lthread_mgr_run(struct lthread_args *args)
{
    struct lthread *lthread_head, *lthread;
    struct wi *wi;
    int i;

    lthread = malloc(sizeof(*lthread));
    memset(lthread, '\0', sizeof(*lthread));
    if (lthread == NULL) {
        fprintf(stderr, "lthread_mgr_run: out of mem\n");
        exit(1);
    }
    lthread->args.listen_addr = args->listen_addr;
    lthread->args.listen_port = args->listen_port;
    lthread->args.wildcard = 1;
    if (lthread_sock_prepare(&lthread->args) != 0) {
        fprintf(stderr, "lthread_sock_prepare(%s:%d) = -1\n", args->listen_addr, args->listen_port);
        exit(1);
    }
    pthread_cond_init(&lthread->args.outpacket_queue.cond, NULL);
    pthread_mutex_init(&lthread->args.outpacket_queue.mutex, NULL);
    lthread->args.outpacket_queue.name = strdup("B2B->NET (wildcard)");
    lthread->args.bslots = args->bslots;

    i = pthread_create(&lthread->rx_thread[0], NULL, (void *(*)(void *))&lthread_rx, &lthread->args);
    printf("%d\n", i);
    lthread_head = lthread;
    for (;;) {
        wi = queue_get_item(&args->outpacket_queue, 0);
        for (lthread = lthread_head; lthread != NULL; lthread = lthread->next) {
            if (lthread->args.wildcard != 0)
                continue;
            if (lthread->args.listen_port != OUTP(wi).local_port ||
              strcmp(lthread->args.listen_addr, OUTP(wi).local_addr) != 0)
                continue;
            break;
        }
        if (lthread == NULL) {
            lthread = malloc(sizeof(*lthread));
            if (lthread == NULL) {
                fprintf(stderr, "lthread_mgr_run: out of mem\n");
                wi_free(wi);
                continue;
            }
            memset(lthread, '\0', sizeof(*lthread));
            lthread->args.listen_addr = strdup(OUTP(wi).local_addr);
            lthread->args.listen_port = OUTP(wi).local_port;
            if (lthread_sock_prepare(&lthread->args) != 0) {
                fprintf(stderr, "lthread_sock_prepare(%s:%d) = -1\n", lthread->args.listen_addr, lthread->args.listen_port);
                wi_free(wi);
                free(lthread->args.listen_addr);
                free(lthread);
                continue;
            }
            pthread_cond_init(&lthread->args.outpacket_queue.cond, NULL);
            pthread_mutex_init(&lthread->args.outpacket_queue.mutex, NULL);
            asprintf(&lthread->args.outpacket_queue.name, "B2B->NET (%s:%d)", lthread->args.listen_addr,
              lthread->args.listen_port);
            lthread->args.bslots = args->bslots;

            for (i = 0; i < 10; i++) {
                pthread_create(&lthread->rx_thread[i], NULL, (void *(*)(void *))&lthread_rx, &lthread->args);
            }
            lthread->next = lthread_head;
            lthread_head = lthread;
        }
        queue_put_item(wi, &lthread->args.outpacket_queue);
    }
}

extern char *strcasestr (__const char *__haystack, __const char *__needle);

static int
extract_call_id(const char *buf, str *call_id)
{
    const char *cp, *cp1;
    int len;

    for (cp = buf;;) {
        cp1 = strcasestr(cp, "call-id:");
        if (cp1 == NULL) {
            cp = strcasestr(cp, "i:");
            if (cp == NULL)
                return -1;
            len = 2;
        } else {
            cp = cp1;
            len = 8;
        }
        if (cp > buf && cp[-1] != '\n' && cp[-1] != '\r') {
            cp += len;
            continue;
        }
        call_id->s = (char *)cp + len;
        while (call_id->s[0] != '\0' && call_id->s[0] != '\r' &&
          call_id->s[0] != '\n' && isspace(call_id->s[0]))
            call_id->s++;
        if (call_id->s[0] == '\0' || call_id->s[0] == '\r' || call_id->s[0] == '\n')
            return -1;
        call_id->len = 0;
        while (call_id->s[call_id->len] != '\0' && call_id->s[call_id->len] != '\r' &&
          call_id->s[call_id->len] != '\n' && !isspace(call_id->s[call_id->len]))
            call_id->len++;
        if (call_id->len == 0)
            return -1;
        return 0;
    }
}

static int
lthread_sock_prepare(struct lthread_args *args)
{
    struct sockaddr_storage ia;
    char listen_port[10];
    int n, on;

    sprintf(listen_port, "%d", args->listen_port);
    n = resolve(sstosa(&ia), AF_INET, args->listen_addr, listen_port, AI_PASSIVE);
    if (n != 0)
        return -1;
    args->sock = socket(AF_INET, SOCK_DGRAM, 0);
    if (args->sock < 0)
        return -1;
    setsockopt(args->sock, SOL_SOCKET, SO_REUSEADDR, &args->sock, sizeof args->sock);
    if (args->wildcard != 0) {
        on = 1;
#if defined(__FreeBSD__)
        setsockopt(args->sock, IPPROTO_IP, IP_RECVDSTADDR, &on, sizeof(on));
#else
	setsockopt(args->sock, SOL_IP, IP_PKTINFO, &on, sizeof(on));
#endif
    }
    n = bind(args->sock, sstosa(&ia), SS_LEN(&ia));
    if (n != 0) {
        printf("lthread_sock_prepare:bind(%d)\n", n);
        close(args->sock);
        return -1;
    }
    return 0;
}

static void
lthread_rx(struct lthread_args *args)
{
    int n;
    socklen_t ralen, lalen;
    size_t rsize;
    struct sockaddr_storage ia, la;
    struct wi *wi;
    pthread_t tx_thread;
    str call_id;
    struct b2bua_slot *bslot;

    if (args->wildcard == 0)
{'fraction_non_alphanumeric': 0.15290924276169265, 'fraction_numerical': 0.009187082405345211, 'mean_word_length': 2.3753817242189337, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 175, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1330690', 'n_tokens_mistral': 5281, 'n_tokens_neox': 5165, 'n_words': 1015}