core/net/rime/collect.c

   1 /**
   2  * \addtogroup rimecollect
   3  * @{
   4  */
   5
   6 /*
   7  * Copyright (c) 2006, Swedish Institute of Computer Science.
   8  * All rights reserved.
   9  *
  10  * Redistribution and use in source and binary forms, with or without
  11  * modification, are permitted provided that the following conditions
  12  * are met:
  13  * 1. Redistributions of source code must retain the above copyright
  14  *    notice, this list of conditions and the following disclaimer.
  15  * 2. Redistributions in binary form must reproduce the above copyright
  16  *    notice, this list of conditions and the following disclaimer in the
  17  *    documentation and/or other materials provided with the distribution.
  18  * 3. Neither the name of the Institute nor the names of its contributors
  19  *    may be used to endorse or promote products derived from this software
  20  *    without specific prior written permission.
  21  *
  22  * THIS SOFTWARE IS PROVIDED BY THE INSTITUTE AND CONTRIBUTORS ``AS IS'' AND
  23  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  24  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  25  * ARE DISCLAIMED.  IN NO EVENT SHALL THE INSTITUTE OR CONTRIBUTORS BE LIABLE
  26  * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  27  * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
  28  * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  29  * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  30  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
  31  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  32  * SUCH DAMAGE.
  33  *
  34  * This file is part of the Contiki operating system.
  35  *
  36  * $Id: collect.c,v 1.36 2010/03/02 15:41:23 dak664 Exp $
  37  */
  38
  39 /**
  40  * \file
  41  *         Tree-based hop-by-hop reliable data collection
  42  * \author
  43  *         Adam Dunkels <adam@sics.se>
  44  */
  45
  46 #include "contiki.h"
  47
  48 #include "net/rime.h"
  49 #include "net/rime/neighbor.h"
  50 #include "net/rime/collect.h"
  51
  52 #include "net/rime/packetqueue.h"
  53
  54 #include "dev/radio-sensor.h"
  55
  56 #include "lib/random.h"
  57
  58 #include <string.h>
  59 #include <stdio.h>
  60 #include <stddef.h>
  61
  62 static const struct packetbuf_attrlist attributes[] =
  63   {
  64     COLLECT_ATTRIBUTES
  65     PACKETBUF_ATTR_LAST
  66   };
  67
  68 #define NUM_RECENT_PACKETS 8
  69
  70 struct recent_packet {
  71   rimeaddr_t originator;
  72   rimeaddr_t sent_to;
  73   uint8_t seqno;
  74 };
  75
  76 struct ack_msg {
  77   uint8_t flags, dummy;
  78   uint16_t rtmetric;
  79 };
  80
  81 #define ACK_FLAGS_CONGESTED         0x80
  82 #define ACK_FLAGS_DROPPED           0x40
  83 #define ACK_FLAGS_LIFETIME_EXCEEDED 0x20
  84
  85 static struct recent_packet recent_packets[NUM_RECENT_PACKETS];
  86 static uint8_t recent_packet_ptr;
  87
  88 #define REXMIT_TIME CLOCK_SECOND * 2
  89 #define FORWARD_PACKET_LIFETIME (6 * (REXMIT_TIME) << 3)
  90 #define MAX_SENDING_QUEUE 6
  91 PACKETQUEUE(sending_queue, MAX_SENDING_QUEUE);
  92
  93 #define SINK 0
  94 #define RTMETRIC_MAX COLLECT_MAX_DEPTH
  95
  96 #define MAX_HOPLIM 10
  97
  98 #ifndef COLLECT_CONF_ANNOUNCEMENTS
  99 #define COLLECT_ANNOUNCEMENTS 0
 100 #else
 101 #define COLLECT_ANNOUNCEMENTS COLLECT_CONF_ANNOUNCEMENTS
 102 #endif /* COLLECT_CONF_ANNOUNCEMENTS */
 103
 104 #define DEBUG 0
 105 #if DEBUG
 106 #include <stdio.h>
 107 #define PRINTF(...) printf(__VA_ARGS__)
 108 #else
 109 #define PRINTF(...)
 110 #endif
 111
 112 #if CONTIKI_TARGET_NETSIM
 113 #include "ether.h"
 114 #endif /* CONTIKI_TARGET_NETSIM */
 115
 116 static void send_queued_packet(void);
 117 static void retransmit_callback(void *ptr);
 118 /*---------------------------------------------------------------------------*/
 119 static void
 120 update_rtmetric(struct collect_conn *tc)
 121 {
 122   struct neighbor *n;
 123
 124   /* We should only update the rtmetric if we are not the sink. */
 125   if(tc->rtmetric != SINK) {
 126
 127     /* Find the neighbor with the lowest rtmetric. */
 128     n = neighbor_best();
 129
 130     /* If n is NULL, we have no best neighbor. */
 131     if(n == NULL) {
 132
 133       /* If we have don't have any neighbors, we set our rtmetric to
 134          the maximum value to indicate that we do not have a route. */
 135
 136       if(tc->rtmetric != RTMETRIC_MAX) {
 137         PRINTF("%d.%d: didn't find a best neighbor, setting rtmetric to max\n",
 138                rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1]);
 139       }
 140       tc->rtmetric = RTMETRIC_MAX;
 141 #if COLLECT_ANNOUNCEMENTS
 142       announcement_set_value(&tc->announcement, tc->rtmetric);
 143 #else /* COLLECT_ANNOUNCEMENTS */
 144       neighbor_discovery_set_val(&tc->neighbor_discovery_conn, tc->rtmetric);
 145 #endif /* COLLECT_ANNOUNCEMENTS */
 146     } else {
 147       /* We set our rtmetric to the rtmetric of our best neighbor plus
 148          the expected transmissions to reach that neighbor. */
 149       if(n->rtmetric + neighbor_etx(n) != tc->rtmetric) {
 150         uint16_t old_rtmetric = tc->rtmetric;
 151
 152         tc->rtmetric = n->rtmetric + neighbor_etx(n);
 153
 154 #if ! COLLECT_ANNOUNCEMENTS
 155
 156         /* If we get a significantly better rtmetric than we had
 157            before, we call neighbor_discovery_start to start a new
 158            period. */
 159         if(old_rtmetric >= tc->rtmetric + NEIGHBOR_ETX_SCALE + NEIGHBOR_ETX_SCALE / 2) {
 160           neighbor_discovery_start(&tc->neighbor_discovery_conn, tc->rtmetric);
 161         } else {
 162           neighbor_discovery_set_val(&tc->neighbor_discovery_conn, tc->rtmetric);
 163         }
 164 #else /* ! COLLECT_ANNOUNCEMENTS */
 165         announcement_set_value(&tc->announcement, tc->rtmetric);
 166 #endif /* ! COLLECT_ANNOUNCEMENTS */
 167
 168         PRINTF("%d.%d: new rtmetric %d\n",
 169                rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1],
 170                tc->rtmetric);
 171
 172         /* We got a new, working, route we send any queued packets we may have. */
 173         if(old_rtmetric == RTMETRIC_MAX) {
 174           send_queued_packet();
 175         }
 176       }
 177     }
 178   }
 179
 180   /*  DEBUG_PRINTF("%d: new rtmetric %d\n", node_id, rtmetric);*/
 181 #if CONTIKI_TARGET_NETSIM
 182   {
 183     char buf[8];
 184     if(tc->rtmetric == RTMETRIC_MAX) {
 185       strcpy(buf, " ");
 186     } else {
 187       sPRINTF(buf, "%.1f", (float)tc->rtmetric / NEIGHBOR_ETX_SCALE);
 188     }
 189     ether_set_text(buf);
 190   }
 191 #endif
 192 }
 193 /*---------------------------------------------------------------------------*/
 194 static void
 195 send_queued_packet(void)
 196 {
 197   struct queuebuf *q;
 198   struct neighbor *n;
 199   struct packetqueue_item *i;
 200   struct collect_conn *c;
 201
 202   PRINTF("%d.%d: send_queued_packet queue len %d\n",
 203          rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1],
 204          packetqueue_len(&sending_queue));
 205
 206   i = packetqueue_first(&sending_queue);
 207   if(i == NULL) {
 208       PRINTF("%d.%d: nothing on queue\n",
 209              rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1]);
 210     /* No packet on the queue, so there is nothing for us to send. */
 211     return;
 212   }
 213   c = packetqueue_ptr(i);
 214   if(c == NULL) {
 215     /* c should not be NULL, but we check it just to be sure. */
 216     PRINTF("%d.%d: queue, c == NULL!\n",
 217              rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1]);
 218     return;
 219   }
 220
 221   if(c->sending) {
 222     /* If we are currently sending a packet, we wait until the
 223        packet is forwarded and try again then. */
 224     PRINTF("%d.%d: queue, c is sending\n",
 225              rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1]);
 226     return;
 227   }
 228
 229   q = packetqueue_queuebuf(i);
 230   if(q != NULL) {
 231     PRINTF("%d.%d: queue, q is on queue\n",
 232            rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1]);
 233     queuebuf_to_packetbuf(q);
 234
 235     n = neighbor_best();
 236
 237     while(n != NULL && rimeaddr_cmp(&n->addr, packetbuf_addr(PACKETBUF_ADDR_SENDER))) {
 238       PRINTF("%d.%d: avoiding fowarding loop to %d.%d\n",
 239              rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1],
 240              n->addr.u8[0], n->addr.u8[1]);
 241       neighbor_remove(&n->addr);
 242       update_rtmetric(c);
 243       n = neighbor_best();;
 244     }
 245
 246     /* Don't send to the neighbor if it is the same neighbor that sent
 247        us the packet. */
 248     if(n != NULL) {
 249       clock_time_t time;
 250       uint8_t rexmit_time_scaling;
 251 #if CONTIKI_TARGET_NETSIM
 252       ether_set_line(n->addr.u8[0], n->addr.u8[1]);
 253 #endif /* CONTIKI_TARGET_NETSIM */
 254       PRINTF("%d.%d: sending packet to %d.%d\n",
 255              rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1],
 256              n->addr.u8[0], n->addr.u8[1]);
 257
 258       rimeaddr_copy(&c->current_receiver, &n->addr);
 259       c->sending = 1;
 260       c->transmissions = 0;
 261       c->max_rexmits = 8;//packetbuf_attr(PACKETBUF_ATTR_EMAX_REXMIT);
 262       PRINTF("max_rexmits %d\n", c->max_rexmits);
 263       packetbuf_set_attr(PACKETBUF_ATTR_RELIABLE, 1);
 264       packetbuf_set_attr(PACKETBUF_ATTR_MAX_MAC_REXMIT, 2);
 265       packetbuf_set_attr(PACKETBUF_ATTR_PACKET_ID, c->seqno);
 266       unicast_send(&c->unicast_conn, &n->addr);
 267       rexmit_time_scaling = c->transmissions;
 268       if(rexmit_time_scaling > 3) {
 269         rexmit_time_scaling = 3;
 270       }
 271       time = REXMIT_TIME << rexmit_time_scaling;
 272       time = time / 2 + random_rand() % (time / 2);
 273       PRINTF("retransmission time %lu\n", time);
 274       ctimer_set(&c->retransmission_timer, time,
 275                  retransmit_callback, c);
 276     } else {
 277     }
 278   }
 279 }
 280 /*---------------------------------------------------------------------------*/
 281 static void
 282 send_next_packet(struct collect_conn *tc)
 283 {
 284   /* Cancel retransmission timer. */
 285   ctimer_stop(&tc->retransmission_timer);
 286
 287   /* Remove the first packet on the queue, the packet that was just sent. */
 288   packetqueue_dequeue(&sending_queue);
 289   tc->seqno = (tc->seqno + 1) % (1 << COLLECT_PACKET_ID_BITS);
 290   tc->sending = 0;
 291   tc->transmissions = 0;
 292
 293   /* Send the next packet in the queue, if any. */
 294   send_queued_packet();
 295 }
 296 /*---------------------------------------------------------------------------*/
 297 static void
 298 handle_ack(struct collect_conn *tc)
 299 {
 300   struct ack_msg *msg;
 301   uint16_t rtmetric;
 302   struct neighbor *n;
 303
 304   if(rimeaddr_cmp(packetbuf_addr(PACKETBUF_ADDR_SENDER),
 305                   &tc->current_receiver) &&
 306      packetbuf_attr(PACKETBUF_ATTR_PACKET_ID) == tc->seqno) {
 307
 308
 309     msg = packetbuf_dataptr();
 310     memcpy(&rtmetric, &msg->rtmetric, sizeof(uint16_t));
 311     n = neighbor_find(packetbuf_addr(PACKETBUF_ADDR_SENDER));
 312     if(n != NULL) {
 313       neighbor_update(n, rtmetric);
 314       update_rtmetric(tc);
 315     }
 316
 317     PRINTF("%d.%d: ACK from %d.%d after %d transmissions, flags %02x\n",
 318            rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1],
 319            tc->current_receiver.u8[0], tc->current_receiver.u8[1],
 320            tc->transmissions,
 321            msg->flags);
 322
 323     if(!(msg->flags & ACK_FLAGS_DROPPED)) {
 324       send_next_packet(tc);
 325     }
 326   }
 327 }
 328 /*---------------------------------------------------------------------------*/
 329 static void
 330 send_ack(struct collect_conn *tc, const rimeaddr_t *to, int congestion, int dropped, int ttl)
 331 {
 332   struct ack_msg *ack;
 333   struct queuebuf *q;
 334   uint16_t packet_seqno;
 335
 336
 337   PRINTF("send_ack\n");
 338
 339   packet_seqno = packetbuf_attr(PACKETBUF_ATTR_PACKET_ID);
 340
 341   q = queuebuf_new_from_packetbuf();
 342   if(q != NULL) {
 343
 344     packetbuf_clear();
 345     packetbuf_set_datalen(sizeof(struct ack_msg));
 346     ack = packetbuf_dataptr();
 347     memset(ack, 0, sizeof(struct ack_msg));
 348     ack->rtmetric = tc->rtmetric;
 349     ack->flags = (congestion? ACK_FLAGS_CONGESTED: 0) |
 350       (dropped? ACK_FLAGS_DROPPED: 0) |
 351       (ttl? ACK_FLAGS_LIFETIME_EXCEEDED: 0);
 352     /* XXX: send explicit congestion notification in ACK queue full; add rtmetric to ACK. */
 353     packetbuf_set_addr(PACKETBUF_ADDR_RECEIVER, to);
 354     packetbuf_set_attr(PACKETBUF_ATTR_PACKET_TYPE, PACKETBUF_ATTR_PACKET_TYPE_ACK);
 355     packetbuf_set_attr(PACKETBUF_ATTR_RELIABLE, 0);
 356     packetbuf_set_attr(PACKETBUF_ATTR_ERELIABLE, 0);
 357     packetbuf_set_attr(PACKETBUF_ATTR_PACKET_ID, packet_seqno);
 358     packetbuf_set_attr(PACKETBUF_ATTR_MAX_REXMIT, 2);
 359     unicast_send(&tc->unicast_conn, to);
 360
 361     PRINTF("%d.%d: collect: Sending ACK to %d.%d for %d\n",
 362            rimeaddr_node_addr.u8[0],rimeaddr_node_addr.u8[1],
 363            to->u8[0],
 364            to->u8[1],
 365            packet_seqno);
 366
 367     RIMESTATS_ADD(acktx);
 368
 369     queuebuf_to_packetbuf(q);
 370     queuebuf_free(q);
 371   } else {
 372     PRINTF("%d.%d: collect: could not send ACK to %d.%d for %d: no queued buffers\n",
 373            rimeaddr_node_addr.u8[0],rimeaddr_node_addr.u8[1],
 374            to->u8[0], to->u8[1],
 375            packet_seqno);
 376   }
 377 }
 378 /*---------------------------------------------------------------------------*/
 379 static void
 380 node_packet_received(struct unicast_conn *c, const rimeaddr_t *from)
 381 {
 382   struct collect_conn *tc = (struct collect_conn *)
 383     ((char *)c - offsetof(struct collect_conn, unicast_conn));
 384   int i;
 385   struct neighbor *n;
 386
 387   /* To protect against sending duplicate packets, we keep a list
 388      of recently forwarded packet seqnos. If the seqno of the current
 389      packet exists in the list, we drop the packet and increase the
 390      ETX of the neighbor we sent it to in the first place. */
 391   if(packetbuf_attr(PACKETBUF_ATTR_PACKET_TYPE) ==
 392      PACKETBUF_ATTR_PACKET_TYPE_DATA) {
 393     rimeaddr_t ack_to;
 394     uint8_t packet_seqno;
 395
 396     rimeaddr_copy(&ack_to, packetbuf_addr(PACKETBUF_ADDR_SENDER));
 397     packet_seqno = packetbuf_attr(PACKETBUF_ATTR_PACKET_ID);
 398
 399     if(rimeaddr_cmp(&tc->last_received_addr, packetbuf_addr(PACKETBUF_ADDR_SENDER)) &&
 400        tc->last_received_seqno == packetbuf_attr(PACKETBUF_ATTR_PACKET_ID)) {
 401       /* This is a duplicate of the packet we last received, so we just send an ACK. */
 402       send_ack(tc, &ack_to, 0, 0, 0);
 403       return;
 404     }
 405     rimeaddr_copy(&tc->last_received_addr, packetbuf_addr(PACKETBUF_ADDR_SENDER));
 406     tc->last_received_seqno = packetbuf_attr(PACKETBUF_ATTR_PACKET_ID);
 407
 408     for(i = 0; i < NUM_RECENT_PACKETS; i++) {
 409       if(recent_packets[i].seqno == packetbuf_attr(PACKETBUF_ATTR_EPACKET_ID) &&
 410          rimeaddr_cmp(&recent_packets[i].originator,
 411                       packetbuf_addr(PACKETBUF_ADDR_ESENDER))) {
 412         PRINTF("%d.%d: found duplicate packet from %d.%d with seqno %d, via %d.%d\n",
 413                rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1],
 414                recent_packets[i].originator.u8[0], recent_packets[i].originator.u8[1],
 415                packetbuf_attr(PACKETBUF_ATTR_EPACKET_ID),
 416                packetbuf_addr(PACKETBUF_ADDR_SENDER)->u8[0],
 417                packetbuf_addr(PACKETBUF_ADDR_SENDER)->u8[1]);
 418         n = neighbor_find(&recent_packets[i].sent_to);
 419         if(n != NULL) {
 420           neighbor_update_etx(n, neighbor_etx(n) / NEIGHBOR_ETX_SCALE + 4);
 421           update_rtmetric(tc);
 422         }
 423         break;
 424       }
 425     }
 426     recent_packets[recent_packet_ptr].seqno = packetbuf_attr(PACKETBUF_ATTR_EPACKET_ID);
 427     rimeaddr_copy(&recent_packets[recent_packet_ptr].originator,
 428                   packetbuf_addr(PACKETBUF_ADDR_ESENDER));
 429     /*    n = neighbor_best();*/
 430
 431     if(tc->rtmetric != SINK) {
 432       n = neighbor_best();
 433       rimeaddr_copy(&recent_packets[recent_packet_ptr].sent_to,
 434                     &n->addr);
 435     } else {
 436       rimeaddr_copy(&recent_packets[recent_packet_ptr].sent_to,
 437                     &rimeaddr_null);
 438     }
 439
 440     recent_packet_ptr = (recent_packet_ptr + 1) % NUM_RECENT_PACKETS;
 441
 442     if(tc->rtmetric == SINK) {
 443
 444       /* If we are the sink, we call the receive function. */
 445
 446       send_ack(tc, &ack_to, 0, 0, 0);
 447
 448       PRINTF("%d.%d: sink received packet %d from %d.%d via %d.%d\n",
 449              rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1],
 450              packetbuf_attr(PACKETBUF_ATTR_EPACKET_ID),
 451              packetbuf_addr(PACKETBUF_ADDR_ESENDER)->u8[0],
 452              packetbuf_addr(PACKETBUF_ADDR_ESENDER)->u8[1],
 453              from->u8[0], from->u8[1]);
 454
 455       if(tc->cb->recv != NULL) {
 456         tc->cb->recv(packetbuf_addr(PACKETBUF_ADDR_ESENDER),
 457                      packetbuf_attr(PACKETBUF_ATTR_EPACKET_ID),
 458                      packetbuf_attr(PACKETBUF_ATTR_HOPS));
 459       }
 460       return;
 461     } else if(packetbuf_attr(PACKETBUF_ATTR_TTL) > 1 &&
 462               tc->rtmetric != RTMETRIC_MAX) {
 463
 464       /* If we are not the sink, we forward the packet to the best
 465          neighbor. */
 466       packetbuf_set_attr(PACKETBUF_ATTR_HOPS, packetbuf_attr(PACKETBUF_ATTR_HOPS) + 1);
 467       packetbuf_set_attr(PACKETBUF_ATTR_TTL, packetbuf_attr(PACKETBUF_ATTR_TTL) - 1);
 468
 469
 470       PRINTF("%d.%d: packet received from %d.%d via %d.%d, sending %d\n",
 471              rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1],
 472              packetbuf_addr(PACKETBUF_ADDR_ESENDER)->u8[0],
 473              packetbuf_addr(PACKETBUF_ADDR_ESENDER)->u8[1],
 474              from->u8[0], from->u8[1], tc->sending);
 475
 476       if(packetqueue_enqueue_packetbuf(&sending_queue, FORWARD_PACKET_LIFETIME,
 477                                        tc)) {
 478         send_ack(tc, &ack_to, 0, 0, 0);
 479         send_queued_packet();
 480       } else {
 481         send_ack(tc, &ack_to, 0, 1, 0);
 482         PRINTF("%d.%d: packet dropped: no queue buffer available\n",
 483                rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1]);
 484       }
 485     } else if(packetbuf_attr(PACKETBUF_ATTR_TTL) <= 1) {
 486       PRINTF("%d.%d: packet dropped: ttl %d\n",
 487              rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1],
 488              packetbuf_attr(PACKETBUF_ATTR_TTL));
 489       send_ack(tc, &ack_to, 0, 1, 1);
 490     }
 491   } else if(packetbuf_attr(PACKETBUF_ATTR_PACKET_TYPE) ==
 492             PACKETBUF_ATTR_PACKET_TYPE_ACK) {
 493     PRINTF("Collect: incoming ack %d from %d.%d (%d.%d) seqno %d (%d)\n",
 494            packetbuf_attr(PACKETBUF_ATTR_PACKET_TYPE),
 495            packetbuf_addr(PACKETBUF_ADDR_SENDER)->u8[0],
 496            packetbuf_addr(PACKETBUF_ADDR_SENDER)->u8[1],
 497            tc->current_receiver.u8[0],
 498            tc->current_receiver.u8[1],
 499            packetbuf_attr(PACKETBUF_ATTR_PACKET_ID),
 500            tc->seqno);
 501     handle_ack(tc);
 502   }
 503   return;
 504 }
 505 /*---------------------------------------------------------------------------*/
 506 static void
 507 node_packet_sent(struct unicast_conn *c, int status, int transmissions)
 508 {
 509   struct collect_conn *tc = (struct collect_conn *)
 510     ((char *)c - offsetof(struct collect_conn, unicast_conn));
 511
 512   /* For data packets, we record the number of transmissions */
 513   if(packetbuf_attr(PACKETBUF_ATTR_PACKET_TYPE) ==
 514      PACKETBUF_ATTR_PACKET_TYPE_DATA) {
 515     PRINTF("%d.%d: sent to %d.%d after %d transmissions\n",
 516            rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1],
 517            tc->current_receiver.u8[0], tc->current_receiver.u8[1],
 518            transmissions);
 519
 520     /*    neighbor_update_etx(neighbor_find(to), transmissions);
 521           update_rtmetric(tc);*/
 522     tc->transmissions += transmissions;
 523
 524     /* Update ETX with the number of transmissions. */
 525     PRINTF("Updating ETX with %d transmissions\n", tc->transmissions);
 526     neighbor_update_etx(neighbor_find(&tc->current_receiver), tc->transmissions);
 527     update_rtmetric(tc);
 528   }
 529 }
 530 /*---------------------------------------------------------------------------*/
 531 static void
 532 timedout(struct collect_conn *tc)
 533 {
 534   PRINTF("%d.%d: timedout after %d retransmissions: packet dropped\n",
 535          rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1], tc->transmissions);
 536
 537   tc->sending = 0;
 538   neighbor_timedout_etx(neighbor_find(&tc->current_receiver), tc->transmissions);
 539   update_rtmetric(tc);
 540
 541   send_next_packet(tc);
 542 }
 543 /*---------------------------------------------------------------------------*/
 544 static void
 545 retransmit_callback(void *ptr)
 546 {
 547   struct collect_conn *c = ptr;
 548
 549   PRINTF("retransmit\n");
 550   if(c->transmissions >= c->max_rexmits) {
 551     timedout(c);
 552   } else {
 553     c->sending = 0;
 554     send_queued_packet();
 555   }
 556 }
 557 /*---------------------------------------------------------------------------*/
 558 #if !COLLECT_ANNOUNCEMENTS
 559 static void
 560 adv_received(struct neighbor_discovery_conn *c, const rimeaddr_t *from,
 561              uint16_t rtmetric)
 562 {
 563   struct collect_conn *tc = (struct collect_conn *)
 564     ((char *)c - offsetof(struct collect_conn, neighbor_discovery_conn));
 565   struct neighbor *n;
 566
 567   n = neighbor_find(from);
 568
 569   if(n == NULL) {
 570     neighbor_add(from, rtmetric, 1);
 571   } else {
 572     neighbor_update(n, rtmetric);
 573     PRINTF("%d.%d: updating neighbor %d.%d, etx %d\n",
 574            rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1],
 575            n->addr.u8[0], n->addr.u8[1], rtmetric);
 576   }
 577
 578   update_rtmetric(tc);
 579 }
 580 #else
 581 static void
 582 received_announcement(struct announcement *a, const rimeaddr_t *from,
 583                       uint16_t id, uint16_t value)
 584 {
 585   struct collect_conn *tc = (struct collect_conn *)
 586     ((char *)a - offsetof(struct collect_conn, announcement));
 587   struct neighbor *n;
 588
 589   n = neighbor_find(from);
 590
 591   if(n == NULL) {
 592     neighbor_add(from, value, 1);
 593     PRINTF("%d.%d: new neighbor %d.%d, etx %d\n",
 594            rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1],
 595            from->u8[0], from->u8[1], value);
 596   } else {
 597     neighbor_update(n, value);
 598     PRINTF("%d.%d: updating neighbor %d.%d, etx %d\n",
 599            rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1],
 600            n->addr.u8[0], n->addr.u8[1], value);
 601   }
 602
 603   update_rtmetric(tc);
 604 }
 605 #endif /* !COLLECT_ANNOUNCEMENTS */
 606 /*---------------------------------------------------------------------------*/
 607 static const struct unicast_callbacks unicast_callbacks = {node_packet_received,
 608                                                            node_packet_sent};
 609 #if !COLLECT_ANNOUNCEMENTS
 610 static const struct neighbor_discovery_callbacks neighbor_discovery_callbacks =
 611   { adv_received, NULL};
 612 #endif /* !COLLECT_ANNOUNCEMENTS */
 613 /*---------------------------------------------------------------------------*/
 614 void
 615 collect_open(struct collect_conn *tc, uint16_t channels,
 616              const struct collect_callbacks *cb)
 617 {
 618 #if !COLLECT_ANNOUNCEMENTS
 619   neighbor_discovery_open(&tc->neighbor_discovery_conn, channels,
 620                           CLOCK_SECOND * 8,
 621                           CLOCK_SECOND * 32,
 622                           (unsigned) (CLOCK_SECOND * 600UL),
 623                           &neighbor_discovery_callbacks);
 624   neighbor_discovery_start(&tc->neighbor_discovery_conn, tc->rtmetric);
 625 #else /* !COLLECT_ANNOUNCEMENTS */
 626   announcement_register(&tc->announcement, channels, tc->rtmetric,
 627                         received_announcement);
 628   announcement_listen(2);
 629 #endif /* !COLLECT_ANNOUNCEMENTS */
 630
 631   unicast_open(&tc->unicast_conn, channels + 1, &unicast_callbacks);
 632   channel_set_attributes(channels + 1, attributes);
 633   tc->rtmetric = RTMETRIC_MAX;
 634   tc->cb = cb;
 635   neighbor_init();
 636   packetqueue_init(&sending_queue);
 637 }
 638 /*---------------------------------------------------------------------------*/
 639 void
 640 collect_close(struct collect_conn *tc)
 641 {
 642 #if COLLECT_ANNOUNCEMENTS
 643   announcement_remove(&tc->announcement);
 644 #else
 645   neighbor_discovery_close(&tc->neighbor_discovery_conn);
 646 #endif /* COLLECT_ANNOUNCEMENTS */
 647   unicast_close(&tc->unicast_conn);
 648 }
 649 /*---------------------------------------------------------------------------*/
 650 void
 651 collect_set_sink(struct collect_conn *tc, int should_be_sink)
 652 {
 653   if(should_be_sink) {
 654     tc->rtmetric = SINK;
 655 #if !COLLECT_ANNOUNCEMENTS
 656     neighbor_discovery_start(&tc->neighbor_discovery_conn, tc->rtmetric);
 657 #endif /* !COLLECT_ANNOUNCEMENTS */
 658   } else {
 659     tc->rtmetric = RTMETRIC_MAX;
 660   }
 661 #if COLLECT_ANNOUNCEMENTS
 662   announcement_set_value(&tc->announcement, tc->rtmetric);
 663 #endif /* COLLECT_ANNOUNCEMENTS */
 664   update_rtmetric(tc);
 665 }
 666 /*---------------------------------------------------------------------------*/
 667 int
 668 collect_send(struct collect_conn *tc, int rexmits)
 669 {
 670   struct neighbor *n;
 671
 672   packetbuf_set_attr(PACKETBUF_ATTR_EPACKET_ID, tc->eseqno++);
 673   packetbuf_set_addr(PACKETBUF_ADDR_ESENDER, &rimeaddr_node_addr);
 674   packetbuf_set_attr(PACKETBUF_ATTR_HOPS, 1);
 675   packetbuf_set_attr(PACKETBUF_ATTR_TTL, MAX_HOPLIM);
 676   packetbuf_set_attr(PACKETBUF_ATTR_MAX_REXMIT, rexmits);
 677
 678   PRINTF("%d.%d: originating packet %d\n",
 679          rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1],
 680          packetbuf_attr(PACKETBUF_ATTR_EPACKET_ID));
 681
 682
 683   PRINTF("rexmit %d\n", rexmits);
 684
 685   if(tc->rtmetric == SINK) {
 686     packetbuf_set_attr(PACKETBUF_ATTR_HOPS, 0);
 687     if(tc->cb->recv != NULL) {
 688       tc->cb->recv(packetbuf_addr(PACKETBUF_ADDR_ESENDER),
 689                    packetbuf_attr(PACKETBUF_ATTR_EPACKET_ID),
 690                    packetbuf_attr(PACKETBUF_ATTR_HOPS));
 691     }
 692     return 1;
 693   } else {
 694     update_rtmetric(tc);
 695     n = neighbor_best();
 696     if(n != NULL) {
 697 #if CONTIKI_TARGET_NETSIM
 698       ether_set_line(n->addr.u8[0], n->addr.u8[1]);
 699 #endif /* CONTIKI_TARGET_NETSIM */
 700       PRINTF("%d.%d: sending to %d.%d\n",
 701              rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1],
 702              n->addr.u8[0], n->addr.u8[1]);
 703       if(packetqueue_enqueue_packetbuf(&sending_queue, FORWARD_PACKET_LIFETIME,
 704                                        tc)) {
 705         send_queued_packet();
 706         return 1;
 707       } else {
 708         PRINTF("%d.%d: drop originated packet: no queuebuf\n",
 709                rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1]);
 710       }
 711     } else {
 712       PRINTF("%d.%d: did not find any neighbor to send to\n",
 713              rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1]);
 714 #if COLLECT_ANNOUNCEMENTS
 715       announcement_listen(1);
 716 #endif /* COLLECT_ANNOUNCEMENTS */
 717       if(packetqueue_enqueue_packetbuf(&sending_queue, FORWARD_PACKET_LIFETIME,
 718                                        tc)) {
 719         return 1;
 720       } else {
 721         PRINTF("%d.%d: drop originated packet: no queuebuf\n",
 722                rimeaddr_node_addr.u8[0], rimeaddr_node_addr.u8[1]);
 723       }
 724     }
 725   }
 726   return 0;
 727 }
 728 /*---------------------------------------------------------------------------*/
 729 int
 730 collect_depth(struct collect_conn *tc)
 731 {
 732   return tc->rtmetric;
 733 }
 734 /*---------------------------------------------------------------------------*/
 735 /** @} */