From c7caec488a68542543bccc050e13ba22cf6c3415 Mon Sep 17 00:00:00 2001 From: Jeremy Lakeman Date: Thu, 7 Feb 2013 15:16:07 +1030 Subject: [PATCH] Don't aggregate packets together on radio links --- constants.h | 4 + overlay_address.c | 3 + overlay_interface.c | 7 +- overlay_packetformats.c | 377 +++++++++++++++++++++------------------- overlay_packetradio.c | 144 ++++++++++++--- overlay_payload.c | 36 +--- overlay_queue.c | 37 ++-- rhizome_packetformats.c | 5 + serval.h | 10 +- 9 files changed, 364 insertions(+), 259 deletions(-) diff --git a/constants.h b/constants.h index 4e0841e7..901dd259 100644 --- a/constants.h +++ b/constants.h @@ -100,6 +100,10 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. #define PAYLOAD_FLAG_CIPHERED (1<<4) #define PAYLOAD_FLAG_SIGNED (1<<5) +// return codes for parsing mdp packet headers +#define HEADER_PROCESS 1 +#define HEADER_FORWARD 2 + // this can be removed once all overlay messages have been turned into mdp payloads #define PAYLOAD_FLAG_LEGACY_TYPE (1<<7) diff --git a/overlay_address.c b/overlay_address.c index 4fe56981..57dca698 100644 --- a/overlay_address.c +++ b/overlay_address.c @@ -355,6 +355,9 @@ int send_please_explain(struct decode_context *context, struct subscriber *sourc else frame->source = my_subscriber; + if (!context->sender) + frame->source_full=1; + frame->source->send_full=1; frame->destination = destination; diff --git a/overlay_interface.c b/overlay_interface.c index bb2fa823..fca2b134 100644 --- a/overlay_interface.c +++ b/overlay_interface.c @@ -683,9 +683,10 @@ overlay_broadcast_ensemble(int interface_number, return WHYF("Cannot send to interface %s as it is down", interface->name); } - if (interface->type==OVERLAY_INTERFACE_PACKETRADIO) { - return overlay_packetradio_tx_packet(interface,recipientaddr,bytes,len); - } else if (interface->fileP) + if (interface->type==OVERLAY_INTERFACE_PACKETRADIO) + return WHYF("Cannot send aggregate packets to packet radios"); + + if (interface->fileP) { struct dummy_packet packet={ diff --git a/overlay_packetformats.c b/overlay_packetformats.c index 782bd994..746d85ce 100644 --- a/overlay_packetformats.c +++ b/overlay_packetformats.c @@ -132,6 +132,165 @@ int overlay_forward_payload(struct overlay_frame *f){ RETURN(0); } +// Parse the mdp envelope header +// may return (HEADER_PROCESS|HEADER_FORWARD) || -1 +int parseMdpPacketHeader(struct decode_context *context, struct overlay_frame *frame, + struct overlay_buffer *buffer, struct subscriber **nexthop){ + int process=1; + int forward=2; + time_ms_t now = gettime_ms(); + + int flags = ob_get(buffer); + if (flags<0) + return WHY("Unable to read flags"); + + if (flags & PAYLOAD_FLAG_SENDER_SAME){ + if (!context->sender) + context->invalid_addresses=1; + frame->source = context->sender; + }else{ + int ret=overlay_address_parse(context, buffer, &frame->source); + if (ret<0) + return WHY("Unable to parse payload source"); + if (!frame->source || frame->source->reachable==REACHABLE_SELF) + process=forward=0; + } + + if (flags & PAYLOAD_FLAG_TO_BROADCAST){ + if (!(flags & PAYLOAD_FLAG_ONE_HOP)){ + if (overlay_broadcast_parse(buffer, &frame->broadcast_id)) + return WHY("Unable to read broadcast address"); + if (overlay_broadcast_drop_check(&frame->broadcast_id)){ + process=forward=0; + if (config.debug.overlayframes) + DEBUGF("Ignoring duplicate broadcast (%s)", alloca_tohex(frame->broadcast_id.id, BROADCAST_LEN)); + } + } + frame->destination=NULL; + }else{ + int ret=overlay_address_parse(context, buffer, &frame->destination); + if (ret<0) + return WHY("Unable to parse payload destination"); + + if (!frame->destination || frame->destination->reachable!=REACHABLE_SELF) + process=0; + + if (!(flags & PAYLOAD_FLAG_ONE_HOP)){ + ret=overlay_address_parse(context, buffer, nexthop); + if (ret<0) + return WHY("Unable to parse payload nexthop"); + + if (!(*nexthop) || (*nexthop)->reachable!=REACHABLE_SELF) + forward=0; + } + } + + if (flags & PAYLOAD_FLAG_ONE_HOP){ + frame->ttl=1; + }else{ + int ttl_qos = ob_get(buffer); + if (ttl_qos<0) + return WHY("Unable to read ttl"); + frame->ttl = ttl_qos & 0x1F; + frame->queue = (ttl_qos >> 5) & 3; + } + frame->ttl--; + if (frame->ttl<=0) + forward=0; + + if (flags & PAYLOAD_FLAG_LEGACY_TYPE){ + frame->type=ob_get(buffer); + if (frame->type<0) + return WHY("Unable to read type"); + }else + frame->type=OF_TYPE_DATA; + + frame->modifiers=flags; + + if (frame->source) + frame->source->last_rx = now; + + // if we can't understand one of the addresses, skip processing the payload + if (context->invalid_addresses){ + + //if (config.debug.overlayframes) + DEBUG("Skipping payload due to unknown addresses"); + return 0; + } + return forward|process; +} + +int parseEnvelopeHeader(struct decode_context *context, struct overlay_interface *interface, + struct sockaddr_in *addr, struct overlay_buffer *buffer){ + + time_ms_t now = gettime_ms(); + + if (ob_get(buffer)!=magic_header[0] || ob_get(buffer)!=magic_header[1]) + return WHY("Packet type not recognised."); + + if (overlay_address_parse(context, buffer, &context->sender)) + return WHY("Unable to parse sender"); + + int packet_flags = ob_get(buffer); + + int sender_interface = 0; + if (packet_flags & PACKET_INTERFACE) + sender_interface = ob_get(buffer); + + if (packet_flags & PACKET_SEQ) + ob_get(buffer); // sequence number, not implemented yet + + if (context->sender){ + // ignore packets that have been reflected back to me + if (context->sender->reachable==REACHABLE_SELF) + return 1; + + context->sender->last_rx = now; + + // TODO probe unicast links when we detect an address change. + + // if this is a dummy announcement for a node that isn't in our routing table + if (context->sender->reachable == REACHABLE_NONE) { + context->sender->interface = interface; + + if (addr) + context->sender->address = *addr; + else + bzero(&context->sender->address, sizeof context->sender->address); + + context->sender->last_probe = 0; + + // assume for the moment, that we can reply with the same packet type + if (packet_flags&PACKET_UNICAST){ + set_reachable(context->sender, REACHABLE_UNICAST|REACHABLE_ASSUMED); + }else{ + set_reachable(context->sender, REACHABLE_BROADCAST|REACHABLE_ASSUMED); + } + } + + /* Note the probe payload must be queued before any SID/SAS request so we can force the packet to have a full sid */ + if (addr && (context->sender->last_probe==0 || now - context->sender->last_probe > interface->tick_ms*10)) + overlay_send_probe(context->sender, *addr, interface, OQ_MESH_MANAGEMENT); + + if ((!(packet_flags&PACKET_UNICAST)) && context->sender->last_acked + interface->tick_ms <= now){ + overlay_route_ack_selfannounce(interface, + context->sender->last_acked>now - 3*interface->tick_ms?context->sender->last_acked:now, + now,sender_interface,context->sender); + + context->sender->last_acked = now; + } + + } + + if (addr){ + if (packet_flags & PACKET_UNICAST) + context->addr=*addr; + else + context->addr=interface->broadcast_address; + } + return 0; +} + int packetOkOverlay(struct overlay_interface *interface,unsigned char *packet, size_t len, int recvttl, struct sockaddr *recvaddr, size_t recvaddrlen) { @@ -199,215 +358,71 @@ int packetOkOverlay(struct overlay_interface *interface,unsigned char *packet, s struct overlay_buffer *b = ob_static(packet, len); ob_limitsize(b, len); - if (ob_get(b)!=magic_header[0] || ob_get(b)!=magic_header[1]){ - ob_free(b); - RETURN(WHY("Packet type not recognised.")); - } - context.interface = f.interface = interface; - if (recvaddr) f.recvaddr = *((struct sockaddr_in *)recvaddr); - else bzero(&f.recvaddr, sizeof f.recvaddr); - + else + bzero(&f.recvaddr, sizeof f.recvaddr); + if (config.debug.overlayframes) DEBUG("Received overlay packet"); - if (overlay_address_parse(&context, b, &context.sender)){ - WHY("Unable to parse sender"); + if (parseEnvelopeHeader(&context, interface, (struct sockaddr_in *)recvaddr, b)<0){ + ob_free(b); + RETURN(-1); } - int packet_flags = ob_get(b); - - int sender_interface = 0; - if (packet_flags & PACKET_INTERFACE) - sender_interface = ob_get(b); - - if (packet_flags & PACKET_SEQ) - ob_get(b); // sequence number, not implemented yet - - if (context.sender){ - - if (context.sender->reachable==REACHABLE_SELF){ - ob_free(b); - RETURN(0); - } - - context.sender->last_rx = now; - - // TODO probe unicast links when we detect an address change. - - // if this is a dummy announcement for a node that isn't in our routing table - if (context.sender->reachable == REACHABLE_NONE) { - context.sender->interface = interface; - context.sender->address = f.recvaddr; - context.sender->last_probe = 0; - - // assume for the moment, that we can reply with the same packet type - if (packet_flags&PACKET_UNICAST){ - set_reachable(context.sender, REACHABLE_UNICAST|REACHABLE_ASSUMED); - }else{ - set_reachable(context.sender, REACHABLE_BROADCAST|REACHABLE_ASSUMED); - } - } - - /* Note the probe payload must be queued before any SID/SAS request so we can force the packet to have a full sid */ - if (context.sender->last_probe==0 || now - context.sender->last_probe > 5000) - overlay_send_probe(context.sender, f.recvaddr, interface, OQ_MESH_MANAGEMENT); - - if ((!(packet_flags&PACKET_UNICAST)) && context.sender->last_acked + interface->tick_ms <= now){ - overlay_route_ack_selfannounce(interface, - context.sender->last_acked>now - 3*interface->tick_ms?context.sender->last_acked:now, - now,sender_interface,context.sender); - - context.sender->last_acked = now; - } - - } - - if (packet_flags & PACKET_UNICAST) - context.addr=f.recvaddr; - else - context.addr=interface->broadcast_address; - while(b->position < b->sizeLimit){ context.invalid_addresses=0; struct subscriber *nexthop=NULL; bzero(f.broadcast_id.id, BROADCAST_LEN); - int process=1; - int forward=1; - int flags = ob_get(b); - if (flags<0){ - WHY("Unable to parse payload flags"); + + int ret = parseMdpPacketHeader(&context, &f, b, &nexthop); + if (ret<0){ + WHY("Header is too short"); break; } - - if (flags & PAYLOAD_FLAG_SENDER_SAME){ - if (!context.sender) - context.invalid_addresses=1; - f.source = context.sender; - }else{ - if (overlay_address_parse(&context, b, &f.source)){ - WHY("Unable to parse payload source"); - break; - } - if (!f.source || f.source->reachable==REACHABLE_SELF) - process=forward=0; - } - if (flags & PAYLOAD_FLAG_TO_BROADCAST){ - if (!(flags & PAYLOAD_FLAG_ONE_HOP)){ - if (overlay_broadcast_parse(b, &f.broadcast_id)){ - WHY("Unable to parse payload broadcast id"); - break; - } - if (overlay_broadcast_drop_check(&f.broadcast_id)){ - process=forward=0; - if (config.debug.overlayframes) - DEBUGF("Ignoring duplicate broadcast (%s)", alloca_tohex(f.broadcast_id.id, BROADCAST_LEN)); - } - } - f.destination=NULL; - }else{ - if (overlay_address_parse(&context, b, &f.destination)){ - WHY("Unable to parse payload destination"); - break; - } - - if (!f.destination || f.destination->reachable!=REACHABLE_SELF){ - process=0; - } - - if (!(flags & PAYLOAD_FLAG_ONE_HOP)){ - if (overlay_address_parse(&context, b, &nexthop)){ - WHY("Unable to parse payload nexthop"); - break; - } - - if (!nexthop || nexthop->reachable!=REACHABLE_SELF){ - forward=0; - } - } - } - - if (flags & PAYLOAD_FLAG_ONE_HOP){ - f.ttl=1; - }else{ - int ttl_qos = ob_get(b); - if (ttl_qos<0){ - WHY("Unable to parse ttl/qos"); - break; - } - f.ttl = ttl_qos & 0x1F; - f.queue = (ttl_qos >> 5) & 3; - } - f.ttl--; - if (f.ttl<=0) - forward=0; - - if (flags & PAYLOAD_FLAG_LEGACY_TYPE){ - f.type=ob_get(b); - if (f.type<0){ - WHY("Unable to parse payload type"); - break; - } - }else - f.type=OF_TYPE_DATA; - - f.modifiers=flags; - // TODO allow for one byte length unsigned int payload_len = ob_get_ui16(b); if (payload_len > ob_remaining(b)){ - WHYF("Unable to parse payload length (%d)", payload_len); + WHYF("Invalid payload length (%d)", payload_len); break; } int next_payload = ob_position(b) + payload_len; - if (f.source) - f.source->last_rx = now; - - // if we can't understand one of the addresses, skip processing the payload - if (context.invalid_addresses){ - if (config.debug.overlayframes) - DEBUG("Skipping payload due to unknown addresses"); - goto next; - } + if (ret!=0){ + if (config.debug.overlayframes){ + DEBUGF("Received payload type %x, len %d", f.type, next_payload - b->position); + DEBUGF("Payload from %s", alloca_tohex_sid(f.source->sid)); + DEBUGF("Payload to %s", (f.destination?alloca_tohex_sid(f.destination->sid):"broadcast")); + if (!is_all_matching(f.broadcast_id.id, BROADCAST_LEN, 0)) + DEBUGF("Broadcast id %s", alloca_tohex(f.broadcast_id.id, BROADCAST_LEN)); + if (nexthop) + DEBUGF("Next hop %s", alloca_tohex_sid(nexthop->sid)); + } - if (config.debug.overlayframes){ - DEBUGF("Received payload type %x, len %d", f.type, next_payload - b->position); - DEBUGF("Payload from %s", alloca_tohex_sid(f.source->sid)); - DEBUGF("Payload to %s", (f.destination?alloca_tohex_sid(f.destination->sid):"broadcast")); - if (!is_all_matching(f.broadcast_id.id, BROADCAST_LEN, 0)) - DEBUGF("Broadcast id %s", alloca_tohex(f.broadcast_id.id, BROADCAST_LEN)); - if (nexthop) - DEBUGF("Next hop %s", alloca_tohex_sid(nexthop->sid)); - } - - if (!process && !forward) - goto next; + f.payload = ob_slice(b, b->position, payload_len); + if (!f.payload){ + // out of memory? + WHY("Unable to slice payload"); + break; + } + // mark the entire payload as having valid data + ob_limitsize(f.payload, payload_len); - f.payload = ob_slice(b, b->position, payload_len); - if (!f.payload){ - WHY("Payload length is longer than remaining packet size"); - break; - } - // mark the entire payload as having valid data - ob_limitsize(f.payload, payload_len); - - // forward payloads that are for someone else or everyone - if (forward){ - overlay_forward_payload(&f); + // forward payloads that are for someone else or everyone + if (ret&HEADER_FORWARD) + overlay_forward_payload(&f); + + // process payloads that are for me or everyone + if (ret&HEADER_PROCESS) + process_incoming_frame(now, interface, &f, &context); + } - // process payloads that are for me or everyone - if (process){ - process_incoming_frame(now, interface, &f, &context); - } - - next: if (f.payload){ ob_free(f.payload); f.payload=NULL; diff --git a/overlay_packetradio.c b/overlay_packetradio.c index c77cf586..ff5a3c54 100644 --- a/overlay_packetradio.c +++ b/overlay_packetradio.c @@ -1,5 +1,8 @@ #include "serval.h" #include "conf.h" +#include "overlay_address.h" +#include "overlay_buffer.h" +#include "overlay_packet.h" #include /* interface decoder states. broadly based on RFC1055 */ @@ -49,12 +52,40 @@ int overlay_packetradio_setup_port(overlay_interface *interface) int overlay_rx_packet_complete(overlay_interface *interface) { if (interface->recv_offset) { - // dispatch received packet - if (packetOkOverlay(interface, interface->rxbuffer, interface->recv_offset, -1, - NULL,0)) { - if (config.debug.packetradio) - WARN("Corrupted or unsupported packet from packet radio interface"); - } + + struct decode_context context; + struct overlay_buffer *buffer=ob_static(interface->rxbuffer, interface->recv_offset); + ob_limitsize(buffer, interface->recv_offset); + struct overlay_frame frame; + struct subscriber *next_hop=NULL; + + bzero(&context, sizeof(struct decode_context)); + bzero(&frame, sizeof(struct overlay_frame)); + + if (parseEnvelopeHeader(&context, interface, NULL, buffer)) + goto end; + + int packetFlags = parseMdpPacketHeader(&context, &frame, buffer, &next_hop); + if (packetFlags<=0) + goto end; + + frame.payload = ob_slice(buffer, ob_position(buffer), ob_remaining(buffer)); + ob_limitsize(frame.payload, ob_remaining(buffer)); + + // forward payloads that are for someone else or everyone + if (packetFlags&HEADER_FORWARD) + overlay_forward_payload(&frame); + + // process payloads that are for me or everyone + if (packetFlags&HEADER_PROCESS) + process_incoming_frame(gettime_ms(), interface, &frame, &context); + + ob_free(frame.payload); + + end: + send_please_explain(&context, my_subscriber, context.sender); + + ob_free(buffer); } interface->recv_offset=0; return 0; @@ -189,11 +220,32 @@ void overlay_packetradio_poll(struct sched_ent *alarm) return ; } -int overlay_packetradio_tx_packet(overlay_interface *interface, - struct sockaddr_in *recipientaddr, - unsigned char *bytes,int len) +static int encode(unsigned char *src, int src_bytes, unsigned char *dst, int dst_len){ + int offset=0; + int i; + for (i=0;idst_len) + return WHY("Buffer overflow while encoding frame"); + + switch(src[i]) { + case SLIP_END: + dst[offset++]=SLIP_ESC; + dst[offset++]=SLIP_ESC_END; + break; + case SLIP_ESC: + dst[offset++]=SLIP_ESC; + dst[offset++]=SLIP_ESC_ESC; + break; + default: + dst[offset++]=src[i]; + } + } + return offset; +} + +int overlay_packetradio_tx_packet(struct overlay_frame *frame) { - if (config.debug.packetradio) DEBUGF("Sending packet of %d bytes",len); /* This is a bit interesting, because we have to deal with RTS/CTS potentially blocking our writing of the packet. @@ -203,41 +255,77 @@ int overlay_packetradio_tx_packet(overlay_interface *interface, We will surround each packet with SLIP END characters, so we should be able to deal with such truncation in a fairly sane manner. */ + overlay_interface *interface=frame->interface; + int interface_number = interface - overlay_interfaces; - if (len>OVERLAY_INTERFACE_RX_BUFFER_SIZE) + if (frame->payload->position>OVERLAY_INTERFACE_RX_BUFFER_SIZE) return WHYF("Not sending over-size packet"); if (interface->tx_bytes_pending>0) return WHYF("Cannot send two packets at the same time"); + struct overlay_buffer *headers=ob_new(); + if (!headers) + return WHY("could not allocate overlay buffer for headers"); + + + struct decode_context context; + bzero(&context, sizeof(struct decode_context)); + + if (frame->source_full) + my_subscriber->send_full=1; + + if (overlay_packet_init_header(&context, headers, NULL, 0, interface_number, 0)) + goto cleanup; + + struct broadcast *broadcast=NULL; + if ((!frame->destination) && !is_all_matching(frame->broadcast_id.id,BROADCAST_LEN,0)) + broadcast = &frame->broadcast_id; + + if (overlay_frame_build_header(&context, headers, + frame->queue, frame->type, + frame->modifiers, frame->ttl, + broadcast, frame->next_hop, + frame->destination, frame->source)) + goto cleanup; + /* Encode packet with SLIP escaping. XXX - Add error correction here also */ unsigned char *buffer = interface->txbuffer; int out_len=0; - int i; buffer[out_len++]=SLIP_END; - for(i=0;ibytes, headers->position, + buffer+out_len, sizeof(interface->txbuffer) - out_len); + if (encoded<0){ + WHY("Ran out of buffer space while encoding headers"); + goto cleanup; + } + + out_len+=encoded; + + encoded=encode(frame->payload->bytes, frame->payload->position, + buffer+out_len, sizeof(interface->txbuffer) - out_len); + if (encoded<0){ + WHY("Ran out of buffer space while encoding payload body"); + goto cleanup; + } + + out_len+=encoded; buffer[out_len++]=SLIP_END; - if (config.debug.packetradio) DEBUGF("Encoded length is %d",out_len); + if (config.debug.packetradio){ + DEBUGF("Encoded length is %d",out_len); + } interface->tx_bytes_pending=out_len; write_buffer(interface); + ob_free(headers); return 0; + +cleanup: + ob_free(headers); + return -1; } diff --git a/overlay_payload.c b/overlay_payload.c index 6a495c43..460c38cf 100644 --- a/overlay_payload.c +++ b/overlay_payload.c @@ -69,8 +69,6 @@ int overlay_frame_build_header(struct decode_context *context, struct overlay_bu if (ob_append_byte(buff, type)) return -1; } - if (ob_append_rfs(buff, 2)) return -1; - return 0; } @@ -82,12 +80,6 @@ int overlay_frame_append_payload(struct decode_context *context, overlay_interfa Will pick a next hop if one has not been chosen. */ - struct overlay_buffer *headers; - - headers=ob_new(); - - if (!headers) return WHY("could not allocate overlay buffer for headers"); - ob_checkpoint(b); if (config.debug.packetconstruction){ @@ -103,41 +95,23 @@ int overlay_frame_append_payload(struct decode_context *context, overlay_interfa if ((!p->destination) && !is_all_matching(p->broadcast_id.id,BROADCAST_LEN,0)){ broadcast = &p->broadcast_id; } - if (overlay_frame_build_header(context, headers, + if (overlay_frame_build_header(context, b, p->queue, p->type, p->modifiers, p->ttl, broadcast, p->next_hop, p->destination, p->source)) goto cleanup; - int hdr_len=headers->position - (headers->var_length_offset +2); - if (config.debug.packetconstruction) - DEBUGF("Patching RFS for actual_len=%d\n",hdr_len + p->payload->position); - - ob_set_ui16(headers,headers->var_length_offset,hdr_len + p->payload->position); - - /* Write payload format plus total length of header bits */ - if (ob_makespace(b,2+headers->position+p->payload->position)) { - /* Not enough space free in output buffer */ - if (config.debug.packetformats) - DEBUGF("Could not make enough space free in output buffer"); - goto cleanup; - } - - /* Package up headers and payload */ - if (ob_append_bytes(b,headers->bytes,headers->position)) { - WHY("could not append header"); - goto cleanup; - } + if (ob_append_ui16(b, ob_position(p->payload))) + goto cleanup; + if (ob_append_bytes(b,p->payload->bytes,p->payload->position)) { WHY("could not append payload"); goto cleanup; } - - ob_free(headers); + return 0; cleanup: - ob_free(headers); ob_rewind(b); return -1; } diff --git a/overlay_queue.c b/overlay_queue.c index c78d8f5e..bf53a9f0 100644 --- a/overlay_queue.c +++ b/overlay_queue.c @@ -419,9 +419,19 @@ overlay_stuff_packet(struct outgoing_packet *packet, overlay_txqueue *queue, tim } if (!packet->buffer){ - // don't transmit on an interface that uses the serial tx buffer - if (frame->interface->tx_bytes_pending>0) - goto skip; + + // TODO, interface flag for this behaviour? + if (frame->interface->type==OVERLAY_INTERFACE_PACKETRADIO){ + // don't transmit if the serial tx buffer has data + if (frame->interface->tx_bytes_pending>0) + goto skip; + + // send packets without aggregating them together + if (overlay_packetradio_tx_packet(frame)) + goto skip; + + goto sent; + } // can we send a packet on this interface now? if (limit_is_allowed(&frame->interface->transfer_limit)) @@ -437,26 +447,27 @@ overlay_stuff_packet(struct outgoing_packet *packet, overlay_txqueue *queue, tim } } - if (config.debug.overlayframes){ - DEBUGF("Sending payload type %x len %d for %s via %s", frame->type, ob_position(frame->payload), - frame->destination?alloca_tohex_sid(frame->destination->sid):"All", - frame->next_hop?alloca_tohex_sid(frame->next_hop->sid):alloca_tohex(frame->broadcast_id.id, BROADCAST_LEN)); - } - if (overlay_frame_append_payload(&packet->context, packet->interface, frame, packet->buffer)){ // payload was not queued goto skip; } + // don't send rhizome adverts if the packet contains a voice payload + if (frame->queue==OQ_ISOCHRONOUS_VOICE) + packet->add_advertisements=0; + + sent: + if (config.debug.overlayframes){ + DEBUGF("Sent payload type %x len %d for %s via %s", frame->type, ob_position(frame->payload), + frame->destination?alloca_tohex_sid(frame->destination->sid):"All", + frame->next_hop?alloca_tohex_sid(frame->next_hop->sid):alloca_tohex(frame->broadcast_id.id, BROADCAST_LEN)); + } + if (frame->destination) frame->destination->last_tx=now; if (frame->next_hop) frame->next_hop->last_tx=now; - // don't send rhizome adverts if the packet contains a voice payload - if (frame->queue==OQ_ISOCHRONOUS_VOICE) - packet->add_advertisements=0; - // mark the payload as sent int keep_payload = 0; diff --git a/rhizome_packetformats.c b/rhizome_packetformats.c index ae551098..c30d9b42 100644 --- a/rhizome_packetformats.c +++ b/rhizome_packetformats.c @@ -209,6 +209,11 @@ int overlay_rhizome_add_advertisements(struct decode_context *context, int inter RETURN(-1); } + if (ob_append_rfs(e, 2)){ + ob_rewind(e); + RETURN(-1); + } + /* Version of rhizome advert block (1 byte): 1 = manifests then BARs, 2 = BARs only, diff --git a/serval.h b/serval.h index c889a0ed..09187411 100644 --- a/serval.h +++ b/serval.h @@ -466,6 +466,12 @@ void insertTransactionInCache(unsigned char *transaction_id); int overlay_forward_payload(struct overlay_frame *f); int packetOkOverlay(struct overlay_interface *interface,unsigned char *packet, size_t len, int recvttl, struct sockaddr *recvaddr, size_t recvaddrlen); +int parseMdpPacketHeader(struct decode_context *context, struct overlay_frame *frame, + struct overlay_buffer *buffer, struct subscriber **nexthop); +int parseEnvelopeHeader(struct decode_context *context, struct overlay_interface *interface, + struct sockaddr_in *addr, struct overlay_buffer *buffer); +int process_incoming_frame(time_ms_t now, struct overlay_interface *interface, + struct overlay_frame *f, struct decode_context *context); int overlay_frame_process(struct overlay_interface *interface, struct overlay_frame *f); int overlay_frame_resolve_addresses(struct overlay_frame *f); @@ -759,9 +765,7 @@ int fd_poll(); void overlay_interface_discover(struct sched_ent *alarm); void overlay_packetradio_poll(struct sched_ent *alarm); int overlay_packetradio_setup_port(overlay_interface *interface); -int overlay_packetradio_tx_packet(overlay_interface *interface, - struct sockaddr_in *recipientaddr, - unsigned char *bytes,int len); +int overlay_packetradio_tx_packet(struct overlay_frame *frame); void overlay_dummy_poll(struct sched_ent *alarm); void overlay_route_tick(struct sched_ent *alarm); void server_config_reload(struct sched_ent *alarm);