Added ordered packet system

Now packets can be grouped up and sent separately, but still be processed
sequentially regardless of how much they get scrambled over the network.
If a packet group isn't received for 15 seconds, all unprocessed packets
in that group will be dropped. Packet ordering is required for level and
area synchronization information when someone joins the level late.

src/pc/network/network.c

@@ -141,6 +141,11 @@ void network_send_to(u8 localIndex, struct Packet* p) {
     // remember reliable packets
     network_remember_reliable(p);
 
+    // set ordered data (MUST BE IMMEDITAELY BEFORE HASING+SENDING)
+    if (p->orderedGroupId != 0) {
+        packet_set_ordered_data(p);
+    }
+
     // save inside packet buffer
     u32 hash = packet_hash(p);
     memcpy(&p->buffer[p->dataLength], &hash, sizeof(u32));
@@ -234,9 +239,10 @@ void network_update(void) {
         gNetworkSystem->update();
     }
 
-    // update reliable packets
+    // update reliable and ordered packets
     if (gNetworkType != NT_NONE) {
         network_update_reliable();
+        packet_ordered_update();
     }
 }
 

src/pc/network/network_player.c

@@ -118,6 +118,7 @@ u8 network_player_connected(enum NetworkPlayerType type, u8 globalIndex) {
         if (gNetworkType == NT_SERVER) {
             gNetworkPlayerServer = gNetworkPlayerLocal;
         }
+        packet_ordered_clear(globalIndex);
         return 0;
     }
 
@@ -155,6 +156,7 @@ u8 network_player_connected(enum NetworkPlayerType type, u8 globalIndex) {
         if (type == NPT_SERVER) { gNetworkPlayerServer = np; }
         else { chat_add_message_ext("player connected", CMT_SYSTEM, get_player_color(np->globalIndex, 0)); }
         LOG_INFO("player connected, local %d, global %d", i, np->globalIndex);
+        packet_ordered_clear(np->globalIndex);
         return i;
     }
 
@@ -187,6 +189,7 @@ u8 network_player_disconnected(u8 globalIndex) {
         for (int j = 0; j < MAX_SYNC_OBJECTS; j++) { gSyncObjects[j].rxEventId[i] = 0; }
         LOG_INFO("player disconnected, local %d, global %d", i, globalIndex);
         chat_add_message_ext("player disconnected", CMT_SYSTEM, get_player_color(globalIndex, 0));
+        packet_ordered_clear(globalIndex);
         return i;
     }
     return UNKNOWN_GLOBAL_INDEX;

src/pc/network/packets/packet.c

@@ -2,43 +2,8 @@
 #include "../network.h"
 #include "pc/debuglog.h"
 
-void packet_receive(struct Packet* p) {
-    u8 packetType = (u8)p->buffer[0];
-
-    // refuse packets from unknown players other than join request
-    if (gNetworkType == NT_SERVER && p->localIndex == UNKNOWN_LOCAL_INDEX && packetType != PACKET_JOIN_REQUEST) {
-        network_send_kick(EKT_CLOSE_CONNECTION);
-        return;
-    }
-
-    // send an ACK if requested
-    network_send_ack(p);
-
-    // check if we've already seen this packet
-    if (p->localIndex != 0 && p->seqId != 0 && gNetworkPlayers[p->localIndex].connected) {
-        struct NetworkPlayer* np = &gNetworkPlayers[p->localIndex];
-        for (int i = 0; i < MAX_RX_SEQ_IDS; i++) {
-            if (np->rxSeqIds[i] == p->seqId) {
-                LOG_INFO("received duplicate packet");
-                return;
-            }
-        }
-        // remember seq id
-        np->rxSeqIds[np->onRxSeqId] = p->seqId;
-        np->onRxSeqId++;
-        if (np->onRxSeqId >= MAX_RX_SEQ_IDS) { np->onRxSeqId = 0; }
-    }
-
-    // check if we should process the packet
-    if (packet_initial_read(p)) {
-        if (gNetworkType == NT_SERVER && p->destGlobalId != PACKET_DESTINATION_BROADCAST && p->destGlobalId != 0 && packetType != PACKET_ACK) {
-            // this packet is meant for someone else
-            struct Packet p2 = { 0 };
-            packet_duplicate(p, &p2);
-            network_send_to(p->destGlobalId, &p2);
-        } else {
-            // process the packet
-            switch (packetType) {
+void packet_process(struct Packet* p) {
+    switch (p->packetType) {
         case PACKET_ACK:                     network_receive_ack(p);                     break;
         case PACKET_PLAYER:                  network_receive_player(p);                  break;
         case PACKET_OBJECT:                  network_receive_object(p);                  break;
@@ -78,6 +43,48 @@ void packet_receive(struct Packet* p) {
         case PACKET_CUSTOM:                  network_receive_custom(p);                  break;
         default: LOG_ERROR("received unknown packet: %d", p->buffer[0]);
     }
+}
+
+void packet_receive(struct Packet* p) {
+    u8 packetType = (u8)p->buffer[0];
+
+    // refuse packets from unknown players other than join request
+    if (gNetworkType == NT_SERVER && p->localIndex == UNKNOWN_LOCAL_INDEX && packetType != PACKET_JOIN_REQUEST) {
+        network_send_kick(EKT_CLOSE_CONNECTION);
+        return;
+    }
+
+    // send an ACK if requested
+    network_send_ack(p);
+
+    // check if we've already seen this packet
+    if (p->localIndex != 0 && p->seqId != 0 && gNetworkPlayers[p->localIndex].connected) {
+        struct NetworkPlayer* np = &gNetworkPlayers[p->localIndex];
+        for (int i = 0; i < MAX_RX_SEQ_IDS; i++) {
+            if (np->rxSeqIds[i] == p->seqId) {
+                LOG_INFO("received duplicate packet");
+                return;
+            }
+        }
+        // remember seq id
+        np->rxSeqIds[np->onRxSeqId] = p->seqId;
+        np->onRxSeqId++;
+        if (np->onRxSeqId >= MAX_RX_SEQ_IDS) { np->onRxSeqId = 0; }
+    }
+
+    // parse the packet without processing the rest
+    if (packet_initial_read(p)) {
+        if (gNetworkType == NT_SERVER && p->destGlobalId != PACKET_DESTINATION_BROADCAST && p->destGlobalId != 0 && packetType != PACKET_ACK) {
+            // this packet is meant for someone else
+            struct Packet p2 = { 0 };
+            packet_duplicate(p, &p2);
+            network_send_to(p->destGlobalId, &p2);
+        } else if (p->orderedGroupId != 0) {
+            // add the packet to the ordered bucket (may process immediately if it's in order)
+            packet_ordered_add(p);
+        } else {
+            // process the packet
+            packet_process(p);
         }
     }
 

src/pc/network/packets/packet.h

@@ -52,6 +52,7 @@ enum PacketType {
 };
 
 struct Packet {
+    enum PacketType packetType;
     u8 localIndex;
     u16 dataLength;
     u16 cursor;
@@ -62,6 +63,9 @@ struct Packet {
     u8 destGlobalId;
     u16 seqId;
     bool sent;
+    u8 orderedFromGlobalId;
+    u8 orderedGroupId;
+    u8 orderedSeqId;
     u8 buffer[PACKET_LENGTH];
 };
 
@@ -71,6 +75,7 @@ enum KickReasonType {
 };
 
 // packet.c
+void packet_process(struct Packet* p);
 void packet_receive(struct Packet* packet);
 
 // packet_read_write.c
@@ -85,6 +90,7 @@ u32 packet_hash(struct Packet* packet);
 bool packet_check_hash(struct Packet* packet);
 void packet_ordered_begin(void);
 void packet_ordered_end(void);
+void packet_set_ordered_data(struct Packet* packet);
 
 // packet_reliable.c
 void network_forget_all_reliable(void);
@@ -93,6 +99,12 @@ void network_receive_ack(struct Packet* p);
 void network_remember_reliable(struct Packet* p);
 void network_update_reliable(void);
 
+// packet_ordered.c
+void packet_ordered_add(struct Packet* p);
+void packet_ordered_clear_table(u8 globalIndex, u8 groupdId);
+void packet_ordered_clear(u8 globalIndex);
+void packet_ordered_update(void);
+
 // packet_player.c
 void network_update_player(void);
 void network_receive_player(struct Packet* p);

src/pc/network/packets/packet_level_area_inform.c

@@ -1,7 +1,7 @@
 #include <stdio.h>
 #include "../network.h"
 #include "level_table.h"
-//#define DISABLE_MODULE_LOG 1
+#define DISABLE_MODULE_LOG 1
 #include "pc/debuglog.h"
 
 void network_send_level_area_inform(struct NetworkPlayer* np) {

src/pc/network/packets/packet_level_respawn_info.c

@@ -10,7 +10,7 @@
 #include "object_fields.h"
 #include "behavior_table.h"
 #include "model_ids.h"
-//#define DISABLE_MODULE_LOG 1
+#define DISABLE_MODULE_LOG 1
 #include "pc/debuglog.h"
 
 #define ERR_COULD_NOT_FIND_OBJECT ((u16)-1)

src/pc/network/packets/packet_ordered.c

@@ -0,0 +1,238 @@
+#include <stdio.h>
+#include "../network.h"
+#define DISABLE_MODULE_LOG 1
+#include "pc/debuglog.h"
+
+#define PACKET_ORDERED_TIMEOUT 15
+
+struct OrderedPacketList {
+    struct Packet p;
+    struct OrderedPacketList* next;
+};
+
+struct OrderedPacketTable {
+    u8 fromGlobalId;
+    u8 groupId;
+    u8 processSeqId;
+    clock_t lastReceived;
+    struct OrderedPacketList* packets;
+    struct OrderedPacketTable* next;
+};
+
+static struct OrderedPacketTable* orderedPacketTable[MAX_PLAYERS] = { 0 };
+
+static void packet_ordered_check_for_processing(struct OrderedPacketTable* opt) {
+    // sanity check
+    assert(opt != NULL);
+
+    struct OrderedPacketList* opl = opt->packets;
+    struct OrderedPacketList* oplLast = opl;
+    struct Packet* p = NULL;
+
+    while (opl != NULL) {
+        if (opt->processSeqId == opl->p.orderedSeqId) {
+            // we found the packet we're supposed to process!
+            p = &opl->p;
+            break;
+        }
+        oplLast = opl;
+        opl = opl->next;
+    }
+
+    // make sure we found the packet
+    if (p == NULL) { return; }
+
+    // process it
+    packet_process(p);
+    LOG_INFO("processed ordered packet (%d, %d, %d)", p->orderedFromGlobalId, p->orderedGroupId, p->orderedSeqId);
+
+    // remove from linked list
+    if (oplLast == opl) {
+        // we processed the head of the list
+        opt->packets = opl->next;
+    } else {
+        // we processed from after the head
+        oplLast->next = opl->next;
+    }
+
+    // deallocate
+    free(opl);
+
+    // find the next one we have to process.
+    opt->processSeqId++;
+    packet_ordered_check_for_processing(opt);
+}
+
+static void packet_ordered_add_to_table(struct OrderedPacketTable* opt, struct Packet* p) {
+    // sanity check
+    assert(opt != NULL);
+    assert(opt->fromGlobalId == p->orderedFromGlobalId);
+    assert(opt->groupId == p->orderedGroupId);
+
+    if (p->orderedSeqId < opt->processSeqId) {
+        // this packet has already been processed!
+        LOG_INFO("this packet has already been processed!");
+        return;
+    }
+
+    if (p->orderedSeqId == opt->processSeqId) {
+        // this is the packet that the table is waiting on, process it!
+        packet_process(p);
+
+        // find the next one we have to process.
+        opt->processSeqId++;
+        packet_ordered_check_for_processing(opt);
+        return;
+    }
+
+    struct OrderedPacketList* opl = opt->packets;
+    struct OrderedPacketList* oplLast = opl;
+
+    // make sure this packet isn't currently in the list
+    while (opl != NULL) {
+        if (opl->p.orderedSeqId == p->orderedSeqId) {
+            // this packet is already in the list!
+            LOG_INFO("this packet is already in the list!");
+            return;
+        }
+        // iterate
+        oplLast = opl;
+        opl = opl->next;
+    }
+
+    // allocate the packet list
+    opl = malloc(sizeof(struct OrderedPacketList));
+    if (oplLast == NULL) {
+        opt->packets = opl;
+    } else {
+        oplLast->next = opl;
+    }
+
+    // copy the packet over to the list
+    memcpy(&opl->p, p, sizeof(struct Packet));
+    opl->next = NULL;
+
+    LOG_INFO("added to list for (%d, %d, %d)", opt->fromGlobalId, opt->groupId, p->orderedSeqId);
+    opt->lastReceived = clock();
+
+    packet_ordered_check_for_processing(opt);
+}
+
+void packet_ordered_add(struct Packet* p) {
+    u8 globalId = p->orderedFromGlobalId;
+    struct OrderedPacketTable* opt = orderedPacketTable[globalId];
+
+    // try to find a ordered packet table for the packet's group
+    struct OrderedPacketTable* optLast = opt;
+    while (opt != NULL) {
+        if (opt->groupId == p->orderedGroupId) {
+            // found a matching group
+            packet_ordered_add_to_table(opt, p);
+            return;
+        }
+        // iterate
+        optLast = opt;
+        opt = opt->next;
+    }
+
+    // could not find a matching group, allocate a ordered packet table
+    opt = malloc(sizeof(struct OrderedPacketTable));
+
+    // put the opt in the right place
+    if (optLast == NULL) {
+        orderedPacketTable[globalId] = opt;
+    } else {
+        optLast->next = opt;
+    }
+
+    // set opt params
+    opt->fromGlobalId = p->orderedFromGlobalId;
+    opt->groupId      = p->orderedGroupId;
+    opt->processSeqId = 1;
+    opt->packets      = NULL;
+    opt->next         = NULL;
+    opt->lastReceived = clock();
+    LOG_INFO("created table for (%d, %d)", opt->fromGlobalId, opt->groupId);
+
+    // add the packet to the table
+    packet_ordered_add_to_table(opt, p);
+}
+
+void packet_ordered_clear_table(u8 globalIndex, u8 groupId) {
+    LOG_INFO("clearing out ordered packet table for %d (%d)", globalIndex, groupId);
+
+    struct OrderedPacketTable* opt = orderedPacketTable[globalIndex];
+    struct OrderedPacketTable* optLast = opt;
+
+    while (opt != NULL) {
+        if (opt->groupId == groupId) {
+            // clear opl of table
+            struct OrderedPacketList* opl = opt->packets;
+            while (opl != NULL) {
+                struct OrderedPacketList* oplNext = opl->next;
+                free(opl);
+                opl = oplNext;
+                LOG_INFO("cleared out opl");
+            }
+
+            // remove from linked list
+            if (optLast == opt) {
+                orderedPacketTable[globalIndex] = opt->next;
+            } else {
+                optLast->next = opt->next;
+            }
+
+            // deallocate table
+            free(opt);
+            LOG_INFO("cleared out opt");
+            return;
+        }
+
+        // goto the next table
+        optLast = opt;
+        opt = opt->next;
+    }
+}
+
+void packet_ordered_clear(u8 globalIndex) {
+    LOG_INFO("clearing out all ordered packet tables for %d", globalIndex);
+    struct OrderedPacketTable* opt = orderedPacketTable[globalIndex];
+
+    while (opt != NULL) {
+        // clear opl of table
+        struct OrderedPacketList* opl = opt->packets;
+        while (opl != NULL) {
+            struct OrderedPacketList* oplNext = opl->next;
+            free(opl);
+            opl = oplNext;
+            LOG_INFO("cleared out opl");
+        }
+
+        // goto next table and free the current one
+        struct OrderedPacketTable* optNext = opt->next;
+        free(opt);
+        opt = optNext;
+        LOG_INFO("cleared out opt");
+    }
+
+    orderedPacketTable[globalIndex] = NULL;
+}
+
+void packet_ordered_update(void) {
+    clock_t currentClock = clock();
+    // check all ordered tables for a time out
+    for (int i = 0; i < MAX_PLAYERS; i++) {
+        struct OrderedPacketTable* opt = orderedPacketTable[i];
+        while (opt != NULL) {
+            struct OrderedPacketTable* optNext = opt->next;
+            float elapsed = (currentClock - opt->lastReceived) / (float)CLOCKS_PER_SEC;
+
+            if (elapsed > PACKET_ORDERED_TIMEOUT) {
+                // too much time has elapsed since we last received a packet for this group, forget the table!
+                packet_ordered_clear_table(i, opt->groupId);
+            }
+
+            opt = optNext;
+        }
+    }
+}

src/pc/network/packets/packet_read_write.c

@@ -1,10 +1,17 @@
 #include "../network.h"
 #include "game/area.h"
+#include "pc/debuglog.h"
 
 #define PACKET_FLAG_BUFFER_OFFSET        3
 #define PACKET_DESTINATION_BUFFER_OFFSET 4
+#define PACKET_ORDERED_SEQ_ID_OFFSET     7
+
+static u16 sNextSeqNum = 1;
+
+static bool sOrderedPackets = false;
+static u8 sCurrentOrderedGroupId = 0;
+static u8 sCurrentOrderedSeqId = 0;
 
-static u16 nextSeqNum = 1;
 void packet_init(struct Packet* packet, enum PacketType packetType, bool reliable, bool levelAreaMustMatch) {
     memset(packet->buffer, 0, PACKET_LENGTH);
     packet->cursor = 0;
@@ -14,15 +21,18 @@ void packet_init(struct Packet* packet, enum PacketType packetType, bool reliabl
     packet->levelAreaMustMatch = levelAreaMustMatch;
     packet->requestBroadcast = false;
     packet->sent = false;
+    packet->orderedFromGlobalId = sOrderedPackets ? gNetworkPlayerLocal->globalIndex : 0;
+    packet->orderedGroupId      = sOrderedPackets ? sCurrentOrderedGroupId : 0;
+    packet->orderedSeqId        = 0;
 
     packet_write(packet, &packetType, sizeof(u8));
 
     // write seq number
     if (reliable) {
-        packet_write(packet, &nextSeqNum, sizeof(u16));
-        packet->seqId = nextSeqNum;
-        nextSeqNum++;
-        if (nextSeqNum == 0) { nextSeqNum++;  }
+        packet_write(packet, &sNextSeqNum, sizeof(u16));
+        packet->seqId = sNextSeqNum;
+        sNextSeqNum++;
+        if (sNextSeqNum == 0) { sNextSeqNum++;  }
     } else {
         u16 nullSeqNum = 0;
         packet_write(packet, &nullSeqNum, sizeof(u16));
@@ -37,6 +47,13 @@ void packet_init(struct Packet* packet, enum PacketType packetType, bool reliabl
     u8 destination = PACKET_DESTINATION_BROADCAST;
     packet_write(packet, &destination, sizeof(u8));
 
+    // write ordered packet information
+    if (sOrderedPackets) {
+        packet_write(packet, &packet->orderedFromGlobalId, sizeof(u8));
+        packet_write(packet, &packet->orderedGroupId,      sizeof(u8));
+        packet_write(packet, &packet->orderedSeqId,        sizeof(u8));
+    }
+
     // write location
     if (levelAreaMustMatch) {
         packet_write(packet, &gCurrCourseNum, sizeof(s16));
@@ -56,13 +73,15 @@ void packet_duplicate(struct Packet* srcPacket, struct Packet* dstPacket) {
     dstPacket->requestBroadcast = srcPacket->requestBroadcast;
     dstPacket->destGlobalId = srcPacket->destGlobalId;
     dstPacket->sent = false;
+    dstPacket->orderedGroupId = srcPacket->orderedGroupId;
+    dstPacket->orderedSeqId = srcPacket->orderedSeqId;
 
     memcpy(&dstPacket->buffer[0], &srcPacket->buffer[0], srcPacket->dataLength);
 
     if (dstPacket->reliable) {
-        dstPacket->seqId = nextSeqNum;
-        nextSeqNum++;
-        if (nextSeqNum == 0) { nextSeqNum++; }
+        dstPacket->seqId = sNextSeqNum;
+        sNextSeqNum++;
+        if (sNextSeqNum == 0) { sNextSeqNum++; }
     }
     memcpy(&dstPacket->buffer[1], &dstPacket->seqId, 2);
 
@@ -74,6 +93,7 @@ void packet_set_flags(struct Packet* packet) {
     u8 flags = 0;
     flags |= SET_BIT(packet->levelAreaMustMatch,  0);
     flags |= SET_BIT(packet->requestBroadcast,    1);
+    flags |= SET_BIT(packet->orderedGroupId != 0, 2);
     packet->buffer[PACKET_FLAG_BUFFER_OFFSET] = flags;
 }
 
@@ -89,11 +109,15 @@ void packet_write(struct Packet* packet, void* data, u16 length) {
 }
 
 u8 packet_initial_read(struct Packet* packet) {
+    // read packet type
+    packet->packetType = packet->buffer[0];
+
     // read packet flags
     u8 flags = 0;
     packet_read(packet, &flags, sizeof(u8));
     packet->levelAreaMustMatch = GET_BIT(flags, 0);
     packet->requestBroadcast   = GET_BIT(flags, 1);
+    bool packetIsOrdered       = GET_BIT(flags, 2);
 
     // read destination
     packet_read(packet, &packet->destGlobalId, sizeof(u8));
@@ -113,6 +137,13 @@ u8 packet_initial_read(struct Packet* packet) {
         if (levelAreaMismatch) { return FALSE; }
     }
 
+    // read ordered packet information
+    if (packetIsOrdered) {
+        packet_read(packet, &packet->orderedFromGlobalId, sizeof(u8));
+        packet_read(packet, &packet->orderedGroupId,      sizeof(u8));
+        packet_read(packet, &packet->orderedSeqId,        sizeof(u8));
+    }
+
     // don't drop packet
     return TRUE;
 }
@@ -142,9 +173,22 @@ bool packet_check_hash(struct Packet* packet) {
 }
 
 void packet_ordered_begin(void) {
-    // TODO: implement ordered packet streams
+    if (sOrderedPackets) { return; }
+    sOrderedPackets = true;
+
+    sCurrentOrderedGroupId++;
+    if (sCurrentOrderedGroupId == 0) { sCurrentOrderedGroupId++; }
+    sCurrentOrderedSeqId = 1;
 }
 
 void packet_ordered_end(void) {
-    // TODO: implement ordered packet streams
+    sOrderedPackets = false;
+    sCurrentOrderedSeqId = 0;
+}
+
+void packet_set_ordered_data(struct Packet* packet) {
+    if (packet->orderedGroupId == 0) { return; }
+    if (packet->orderedSeqId != 0) { return; }
+    packet->orderedSeqId = sCurrentOrderedSeqId++;
+    packet->buffer[PACKET_ORDERED_SEQ_ID_OFFSET] = packet->orderedSeqId;
 }