X-Git-Url: https://git.rrq.au/?a=blobdiff_plain;f=rrqnet.c;h=69c363c0f9c30bf144a069520cf85270ea919a72;hb=refs%2Ftags%2F1.6.2;hp=8f09381d8c4e071c7a4c3061f68fd4e8eb4f6d78;hpb=aa26cff1836f2c7958893f60f9ffb34e2806c833;p=rrq%2Frrqnet.git diff --git a/rrqnet.c b/rrqnet.c index 8f09381..69c363c 100644 --- a/rrqnet.c +++ b/rrqnet.c @@ -16,6 +16,7 @@ #include #include +#include #include #include #include @@ -99,8 +100,7 @@ typedef struct _ReaderData { } ReaderData; // heartbeat interval, in seconds -#define HEARTBEAT 30 -#define HEARTBEAT_MICROS ( HEARTBEAT * 1000000 ) +#define HEARTBEAT_MICROS ( heart_rate * 1000000 ) // Macros for timing, for struct timeval variables #define TIME_MICROS(TM) (((int64_t) (TM)->tv_sec * 1000000) + (TM)->tv_usec ) @@ -184,6 +184,7 @@ static int udp_fd; static int udp_port; static int threads_count = 0; static int buffers_count = 0; +static int heart_rate = 30; // Setup for multicast channel static struct { @@ -691,6 +692,14 @@ static int parse_threads_count(char *arg) { return 0; } +static int parse_heartbeat_rate(char *arg) { + if ( ( sscanf( arg, "%u", &heart_rate ) != 1 ) || heart_rate < 0 ) { + return 1; + } + VERBOSEOUT( "** Heartbeat rate = %d\n", heart_rate ); + return 0; +} + static int parse_buffers_count(char *arg) { if ( ( sscanf( arg, "%u", &buffers_count ) != 1 ) || buffers_count < 1 ) { return 1; @@ -946,35 +955,63 @@ static void sendpacket4(unsigned char *buf, int n,struct Remote *r) { .msg_controllen = CMSG_SPACE( sizeof( struct in_pktinfo ) ), .msg_flags = 0 // unused }; - if ( r->laddr.in.sa_family && 0 ) { - msg.msg_control = &control; - msg.msg_controllen = CMSG_SPACE( sizeof( struct in_pktinfo ) ); - } - VERBOSE2OUT( "sendmsg %lu from %s to %s\n", + VERBOSE2OUT( "sendmsg ipv4 %zu from %s to %s\n", msg.msg_controllen, inet_stoa( &r->laddr ), inet_stoa( &r->uaddr ) ); if ( sendmsg( udp_fd, &msg, 0 ) < n ) { perror( "Writing socket" ); } -#if 0 // ILLUSTRATION - struct cmsghdr *cmsg = CMSG_FIRSTHDR(&msg); - cmsg->cmsg_level = IPPROTO_IP; - cmsg->cmsg_type = IP_PKTINFO; - cmsg->cmsg_len = CMSG_LEN(sizeof(struct in_pktinfo)); - struct in_pktinfo *pktinfo = (struct in_pktinfo*) CMSG_DATA(cmsg); - pktinfo->ipi_ifindex = src_interface_index; - pktinfo->ipi_spec_dst = src_addr; - msg.msg_controllen = - return sendmsg( fd, &msg, IP_PKTINFO ); -#endif } static void sendpacket6(unsigned char *buf, int n,struct Remote *r) { - //NYI - (void)buf; - (void)n; - (void)r; + // The UDP socket is ipv6, possibly with mapped ipv4 address(es) + struct iovec data[1] = {{ .iov_base = buf, .iov_len = n }}; + struct { + struct cmsghdr hdr; + union { + struct in_pktinfo in4; + struct in6_pktinfo in6; + } data; + } control; + struct msghdr msg = { + .msg_name = &r->uaddr.in6, + .msg_namelen = sizeof( struct sockaddr_in6 ), + .msg_iov = data, + .msg_iovlen = 1, + .msg_control = 0, + .msg_controllen = 0, + .msg_flags = 0 // unused + }; + if ( r->ifindex ) { + switch ( r->laddr.in.sa_family ) { + case AF_INET6: + control.hdr.cmsg_len = CMSG_LEN(sizeof(struct in6_pktinfo)); + control.hdr.cmsg_level = IPPROTO_IPV6; + control.hdr.cmsg_type = IPV6_PKTINFO; + control.data.in6.ipi6_ifindex = r->ifindex; + memcpy( &control.data.in6.ipi6_addr, &r->laddr.in6.sin6_addr, 16 ); + msg.msg_control = &control; + msg.msg_controllen = CMSG_SPACE( sizeof( struct in6_pktinfo ) ); + break; + case AF_INET: + control.hdr.cmsg_len = CMSG_LEN(sizeof(struct in_pktinfo)); + control.hdr.cmsg_level = IPPROTO_IP; + control.hdr.cmsg_type = IP_PKTINFO; + control.data.in4.ipi_ifindex = r->ifindex; + control.data.in4.ipi_spec_dst = r->laddr.in4.sin_addr; + msg.msg_control = &control; + msg.msg_controllen = CMSG_SPACE( sizeof( struct in_pktinfo ) ); + break; + } + } + VERBOSE2OUT( "sendmsg ipv6 %d from %s to %s\n", + r->ifindex, + inet_stoa( &r->laddr ), + inet_stoa( &r->uaddr ) ); + if ( sendmsg( udp_fd, &msg, 0 ) < n ) { + perror( "Writing socket" ); + } } // Write a packet via the given Interface with encryption as specified. @@ -1058,10 +1095,12 @@ static void unmap_if_mapped(struct SockAddr *s) { } // Route the packet from the given src -static struct Interface *input_check( - unsigned char *buf,ssize_t len,struct SockAddr *src ) -{ - VERBOSE2OUT( "RECV %ld bytes from %s\n", len, inet_stoa( src ) ); +static struct Interface *input_check(PacketItem *pi) { + unsigned char *buf = pi->buffer; + ssize_t len = pi->len; + struct SockAddr *src = &pi->src; + + VERBOSE2OUT( "RECV %zd bytes from %s\n", len, inet_stoa( src ) ); struct Remote *r = 0; struct timeval now = { 0 }; if ( gettimeofday( &now, 0 ) ) { @@ -1078,12 +1117,30 @@ static struct Interface *input_check( VERBOSEOUT( "New remote %s by %s\n", inet_stoa( src ), a->source ); r = add_remote( src, a ); //r->rec_when = now; // Set activity stamp of new remote + // Set the local addressing for the remote, unless set already + // Note: potential for multi-thread competition here + if ( udp6 ) { + r->ifindex = pi->dstinfo.in6.ipi6_ifindex; + r->laddr.in6.sin6_family = AF_INET6; + r->laddr.in6.sin6_port = htons( udp_port ); + memcpy( &r->laddr.in6.sin6_addr, + &pi->dstinfo.in6.ipi6_addr, + 16 ); + unmap_if_mapped( &r->laddr ); + } else { + r->ifindex = pi->dstinfo.in4.ipi_ifindex; + r->laddr.in4.sin_family = AF_INET; + r->laddr.in4.sin_port = htons( udp_port ); + memcpy( &r->laddr.in4.sin_addr, + &pi->dstinfo.in4.ipi_spec_dst, + 4 ); + } } if ( len < 12 ) { // Ignore short data, but maintain channel r->rec_when = now; // Update activity stamp touched remote if ( len > 0 ) { - VERBOSEOUT( "Ignoring %ld bytes from %s\n", + VERBOSEOUT( "Ignoring %zd bytes from %s\n", len, inet_stoa( src ) ); } return 0; @@ -1149,12 +1206,12 @@ static struct Interface *input_check( // primary channel, or the time since the last packet for that // interface is less than RECENT_MICROS, with different limits // for broadcast and unicast. - int64_t dmac = DIFF_MICROS( &now, &x->rec_when); + uint64_t dmac = DIFF_MICROS( &now, &x->rec_when); if ( x->remote->spec == 0 || RECENT_MICROS( *buf & 1, dmac ) ) { if ( verbose >= 2 ) { fprintf( stderr, - "Dropped. MAC %s (%ld) from %s, should be %s\n", + "Dropped. MAC %s (%"PRIu64") from %s, should be %s\n", inet_mtoa( buf+6 ), dmac, inet_stoa( src ), inet_stoa( &x->remote->uaddr ) ); } @@ -1180,52 +1237,35 @@ static struct Interface *input_check( // Check packet and deliver out static void route_packet(PacketItem *pi) { - unsigned char *buf = pi->buffer; - int len = pi->len; - struct SockAddr *src = &pi->src; - struct Interface *x = input_check( buf, len, src ); + //unsigned char *buf = pi->buffer; + //ssize_t len = pi->len; + struct Interface *x = input_check( pi ); if ( x == 0 ) { VERBOSE2OUT( "not a nice packet\n" ); return; // not a nice packet } - // Set the local addressing for the remote - if ( udp6 ) { - x->remote->ifindex = pi->dstinfo.in6.ipi6_ifindex; - x->remote->laddr.in6.sin6_family = AF_INET6; - x->remote->laddr.in6.sin6_port = htons( udp_port ); - memcpy( &x->remote->laddr.in6.sin6_addr, - &pi->dstinfo.in6.ipi6_addr, - 16 ); - } else { - x->remote->ifindex = pi->dstinfo.in4.ipi_ifindex; - x->remote->laddr.in4.sin_family = AF_INET; - x->remote->laddr.in4.sin_port = htons( udp_port ); - memcpy( &x->remote->laddr.in4.sin_addr, - &pi->dstinfo.in4.ipi_spec_dst, - 4 ); - } - if ( ( *buf & 1 ) == 0 ) { + if ( ( *pi->buffer & 1 ) == 0 ) { // unicast struct Interface *y = 0; // reuse for destination interface - Interface_FIND( buf, y ); + Interface_FIND( pi->buffer, y ); if ( y == 0 ) { VERBOSE2OUT( "RECV %s -> %s from %s without channel and dropped\n", - inet_mtoa( buf+6 ), inet_mtoa( buf ), + inet_mtoa( pi->buffer + 6 ), inet_mtoa( pi->buffer ), inet_stoa( &x->remote->uaddr ) ); return; } if ( x->remote == y->remote ) { VERBOSEOUT( "RECV loop for %s -> %s from %s to %s\n", - inet_mtoa( buf+6 ), inet_mtoa( buf ), + inet_mtoa( pi->buffer+6 ), inet_mtoa( pi->buffer ), inet_stoa( &x->remote->uaddr ), inet_stoa( &y->remote->uaddr ) ); Interface_DEL( y ); // Need to see this interface again return; } VERBOSE2OUT( "RECV route %s -> %s\n", - inet_mtoa( buf+6 ), - inet_mtoa( buf ) ); - write_remote( buf, len, y->remote ); + inet_mtoa( pi->buffer+6 ), + inet_mtoa( pi->buffer ) ); + write_remote( pi->buffer, pi->len, y->remote ); return; } // broadcast. +x+ is source interface @@ -1236,7 +1276,7 @@ static void route_packet(PacketItem *pi) { now.tv_sec = time( 0 ); } VERBOSE2OUT( "BC %s -> %s from %s to %s\n", - inet_mtoa( buf+6 ), inet_mtoa( buf ), + inet_mtoa( pi->buffer+6 ), inet_mtoa( pi->buffer ), inet_stoa( &x->remote->uaddr ), inet_stoa( &x->remote->laddr ) ); struct Remote *r; @@ -1256,7 +1296,7 @@ static void route_packet(PacketItem *pi) { if ( r->spec && ! is_uplink( r->spec ) && DIFF_MICROS( &now, &r->rec_when ) > VERYOLD_MICROS ) { // remove old downlink connection - VERBOSEOUT( "Old remote discarded %s (%ld)\n", + VERBOSEOUT( "Old remote discarded %s (%"PRId64")\n", inet_stoa( &r->uaddr ), TIME_MICROS( &r->rec_when ) ); // Removing a downlink might have threading implications @@ -1265,7 +1305,7 @@ static void route_packet(PacketItem *pi) { } // Send packet to the remote // Only no-clash or to the tap/stdin - write_remote( buf, len, r ); + write_remote( pi->buffer, pi->len, r ); } Remote_UNLOCK; } @@ -1430,7 +1470,7 @@ static void heartbeat(int fd) { r = (struct Remote *) tmp; VERBOSE3OUT( "heartbeat check %s\n", inet_stoa( &r->uaddr ) ); if ( r->spec && is_uplink( r->spec ) ) { - if ( DIFF_MICROS( &now, &r->rec_when ) > HEARTBEAT_MICROS ) { + if ( DIFF_MICROS( &now, &r->rec_when ) >= HEARTBEAT_MICROS ) { VERBOSE3OUT( "heartbeat %s\n", inet_stoa( &r->uaddr ) ); write_remote( data, 0, r ); } @@ -1595,6 +1635,15 @@ int main(int argc, char *argv[]) { i += 2; ENSUREARGS( 1 ); } + // then: optional -H seconds + if ( strncmp( "-H", argv[i], 2 ) == 0 ) { + ENSUREARGS( 2 ); + if ( parse_heartbeat_rate( argv[i+1] ) ) { + usage(); + } + i += 2; + ENSUREARGS( 1 ); + } // then: optional -m mcast if ( strncmp( "-m", argv[i], 2 ) == 0 ) { ENSUREARGS( 2 ); @@ -1760,8 +1809,12 @@ int main(int argc, char *argv[]) { // Start heartbeating to uplinks for ( ;; ) { - sleep( HEARTBEAT ); - heartbeat( udp_fd ); + if ( heart_rate != 0 ) { + sleep( heart_rate ); + heartbeat( udp_fd ); + } else { + sleep( 600 ); + } } return 0; }