Fix P2P receiver constantly flooding network with request inventory packets

build.gradle

@@ -86,17 +86,17 @@ dependencies {
 	implementation project(":kiwi-java")
 	include project(":kiwi-java")
 
-	modRuntimeOnly "de.siphalor:mousewheelie-1.18:${project.mousewheelie_version}"
-	runtimeOnly(project(":plugin:mousewheelie")) {
-		transitive = false
-	}
-	include project(":plugin:mousewheelie")
-
-	modRuntimeOnly "me.shedaniel:RoughlyEnoughItems-fabric:${project.rei_version}"
-	runtimeOnly(project(":plugin:rei")) {
-		transitive = false
-	}
-	include project(":plugin:rei")
+//	modRuntimeOnly "de.siphalor:mousewheelie-1.18:${project.mousewheelie_version}"
+//	runtimeOnly(project(":plugin:mousewheelie")) {
+//		transitive = false
+//	}
+//	include project(":plugin:mousewheelie")
+//
+//	modRuntimeOnly "me.shedaniel:RoughlyEnoughItems-fabric:${project.rei_version}"
+//	runtimeOnly(project(":plugin:rei")) {
+//		transitive = false
+//	}
+//	include project(":plugin:rei")
 
 	// runtimeOnly project(":plugin:techreborn")
 	// include project(":plugin:techreborn")

kiwi-java

@@ -1 +1 @@
-Subproject commit abe783ae7ce45ccf296ed7b7e5c1b194e7c560fa
+Subproject commit 1cbaea53d207f1e16c6e5ee2e6bf6e3c1440ac44

src/main/java/net/shadowfacts/phycon/api/Interface.java

@@ -16,4 +16,6 @@ public interface Interface {
 
 	void send(@NotNull EthernetFrame frame);
 
+	default void cableDisconnected() {}
+
 }

src/main/kotlin/net/shadowfacts/phycon/block/DeviceBlockEntity.kt

@@ -19,8 +19,10 @@ import net.shadowfacts.phycon.api.util.MACAddress
 import net.shadowfacts.phycon.frame.ARPQueryFrame
 import net.shadowfacts.phycon.frame.ARPResponseFrame
 import net.shadowfacts.phycon.frame.BasePacketFrame
+import net.shadowfacts.phycon.frame.NetworkSplitFrame
 import net.shadowfacts.phycon.packet.*
 import net.shadowfacts.phycon.util.NetworkUtil
+import java.lang.ref.WeakReference
 import java.util.*
 
 /**
@@ -43,6 +45,7 @@ abstract class DeviceBlockEntity(type: BlockEntityType<*>, pos: BlockPos, state:
 
 	private val arpTable = mutableMapOf<IPAddress, MACAddress>()
 	private val packetQueue = LinkedList<PendingPacket>()
+	private var cachedDestination: WeakReference<Interface>? = null
 
 	var counter: Long = 0
 
@@ -73,6 +76,7 @@ abstract class DeviceBlockEntity(type: BlockEntityType<*>, pos: BlockPos, state:
 		when (frame) {
 			is ARPQueryFrame -> handleARPQuery(frame)
 			is ARPResponseFrame -> handleARPResponse(frame)
+			is NetworkSplitFrame -> handleNetworkSplit()
 			is PacketFrame -> {
 				if (frame.packet.destination.isBroadcast || frame.packet.destination == ipAddress) {
 					doHandlePacket(frame.packet)
@@ -105,6 +109,11 @@ abstract class DeviceBlockEntity(type: BlockEntityType<*>, pos: BlockPos, state:
 		packetQueue.removeAll(toRemove)
 	}
 
+	protected open fun handleNetworkSplit() {
+		arpTable.clear()
+		cachedDestination = null
+	}
+
 	override fun sendPacket(packet: Packet) {
 		if (packet.destination.isBroadcast) {
 			send(BasePacketFrame(packet, macAddress, MACAddress.BROADCAST))
@@ -119,14 +128,28 @@ abstract class DeviceBlockEntity(type: BlockEntityType<*>, pos: BlockPos, state:
 	}
 
 	open fun findDestination(): Interface? {
+		val cachedDestination = this.cachedDestination?.get()
+		if (cachedDestination != null) {
+			return cachedDestination
+		}
+
 		val sides = (cachedState.block as NetworkComponentBlock).getNetworkConnectedSides(cachedState, world!!, pos)
 		return when (sides.size) {
 			0 -> null
-			1 -> NetworkUtil.findConnectedInterface(world!!, pos, sides.first())
+			1 -> {
+				NetworkUtil.findConnectedInterface(world!!, pos, sides.first())?.also {
+					this.cachedDestination = WeakReference(it)
+				}
+			}
 			else -> throw RuntimeException("DeviceBlockEntity.findDestination must be overridden by devices which have more than 1 network connected side")
 		}
 	}
 
+	override fun cableDisconnected() {
+		cachedDestination = null
+		handleNetworkSplit()
+	}
+
 	open fun tick() {
 		counter++
 

src/main/kotlin/net/shadowfacts/phycon/block/cable/CableBlock.kt

@@ -2,8 +2,10 @@ package net.shadowfacts.phycon.block.cable
 
 import net.fabricmc.fabric.api.`object`.builder.v1.block.FabricBlockSettings
 import net.minecraft.block.*
+import net.minecraft.block.entity.BlockEntity
 import net.minecraft.entity.player.PlayerEntity
 import net.minecraft.item.ItemPlacementContext
+import net.minecraft.item.ItemStack
 import net.minecraft.state.StateManager
 import net.minecraft.state.property.EnumProperty
 import net.minecraft.util.ActionResult
@@ -23,9 +25,11 @@ import net.shadowfacts.phycon.PhysicalConnectivity
 import net.shadowfacts.phycon.api.Interface
 import net.shadowfacts.phycon.api.NetworkCableBlock
 import net.shadowfacts.phycon.api.NetworkComponentBlock
+import net.shadowfacts.phycon.block.DeviceBlockEntity
 import net.shadowfacts.phycon.init.PhyItems
 import net.shadowfacts.phycon.item.FaceDeviceBlockItem
 import net.shadowfacts.phycon.util.CableConnection
+import net.shadowfacts.phycon.util.NetworkUtil
 import net.shadowfacts.phycon.util.containsInclusive
 import java.util.*
 
@@ -195,4 +199,18 @@ class CableBlock(
 		return getShape(state)
 	}
 
+	override fun onBreak(world: World, pos: BlockPos, state: BlockState, player: PlayerEntity) {
+		super.onBreak(world, pos, state, player)
+		if (!world.isClient) {
+			world.server?.execute {
+				// notify devices on either end that the connection was broken (i.e., unset the cached receivers)
+				val connectedSides = getNetworkConnectedSides(state, world, pos)
+				for (side in connectedSides) {
+					val dest = NetworkUtil.findConnectedInterface(world, pos, side)
+					dest?.cableDisconnected()
+				}
+			}
+		}
+	}
+
 }

src/main/kotlin/net/shadowfacts/phycon/block/netinterface/InterfaceBlockEntity.kt

@@ -19,6 +19,7 @@ import net.shadowfacts.phycon.component.NetworkStackReceiver
 import net.shadowfacts.phycon.component.handleItemStack
 import net.shadowfacts.phycon.packet.*
 import net.shadowfacts.phycon.util.ClientConfigurableDevice
+import java.lang.ref.WeakReference
 import kotlin.math.min
 
 /**
@@ -37,20 +38,21 @@ class InterfaceBlockEntity(pos: BlockPos, state: BlockState): DeviceBlockEntity(
 	override var receiverPriority = 0
 	var syncPriorities = true
 
-	// todo: should this be a weak ref?
-	private var inventory: GroupedItemInv? = null
+	private var inventory: WeakReference<GroupedItemInv>? = null
 
 	fun updateInventory() {
 		val offsetPos = pos.offset(facing)
 		val option = SearchOptions.inDirection(facing)
-		inventory = ItemAttributes.GROUPED_INV.getFirstOrNull(world, offsetPos, option)
+		inventory = ItemAttributes.GROUPED_INV.getFirstOrNull(world, offsetPos, option).let {
+			WeakReference(it)
+		}
 	}
 
 	private fun getInventory(): GroupedItemInv? {
 		// if we don't have an inventory, try to get one
 		// this happens when readAll is called before a neighbor state changes, such as immediately after world load
-		if (inventory == null) updateInventory()
-		return inventory
+		if (inventory?.get() == null) updateInventory()
+		return inventory?.get()
 	}
 
 	override fun handle(packet: Packet) {

src/main/kotlin/net/shadowfacts/phycon/block/netswitch/SwitchBlockEntity.kt

@@ -17,6 +17,7 @@ import net.shadowfacts.phycon.api.frame.PacketFrame
 import net.shadowfacts.phycon.api.util.IPAddress
 import net.shadowfacts.phycon.api.util.MACAddress
 import net.shadowfacts.phycon.frame.BasePacketFrame
+import net.shadowfacts.phycon.frame.NetworkSplitFrame
 import net.shadowfacts.phycon.init.PhyBlockEntities
 import net.shadowfacts.phycon.packet.ItemStackPacket
 import net.shadowfacts.phycon.util.NetworkUtil
@@ -36,6 +37,7 @@ class SwitchBlockEntity(pos: BlockPos, state: BlockState): BlockEntity(PhyBlockE
 	val interfaces = Direction.values().map { SwitchInterface(it, WeakReference(this), MACAddress.random()) }
 
 	private val macTable = mutableMapOf<MACAddress, Direction>()
+	private val destinationCache = Array<WeakReference<Interface>?>(6) { null }
 	private var packetsHandledThisTick = 0
 	private var delayedPackets: Deque<Pair<PacketFrame, SwitchInterface>> = LinkedList()
 
@@ -65,12 +67,12 @@ class SwitchBlockEntity(pos: BlockPos, state: BlockState): BlockEntity(PhyBlockE
 			PhysicalConnectivity.NETWORK_LOGGER.debug("{} ({}, {}) forwarding {} to side {}", this, fromItf.side, fromItf.macAddress, frame, dir)
 			interfaceForSide(dir).send(frame)
 		} else {
-			PhysicalConnectivity.NETWORK_LOGGER.debug("{} ({}, {}) flooding {}", this, fromItf.side, fromItf.macAddress, frame)
 			flood(frame, fromItf)
 		}
 	}
 
-	private fun flood(frame: EthernetFrame, source: Interface) {
+	private fun flood(frame: EthernetFrame, source: SwitchInterface) {
+		PhysicalConnectivity.NETWORK_LOGGER.debug("{} ({}, {}) flooding {}", this, source.side, source.macAddress, frame)
 		for (itf in interfaces) {
 			if (source == itf) continue
 			itf.send(frame)
@@ -79,7 +81,20 @@ class SwitchBlockEntity(pos: BlockPos, state: BlockState): BlockEntity(PhyBlockE
 
 	private fun findDestination(fromItf: Interface): Interface? {
 		val side = (fromItf as SwitchInterface).side
-		return NetworkUtil.findConnectedInterface(world!!, pos, side)
+		return destinationCache[side.ordinal]?.get()
+				?: NetworkUtil.findConnectedInterface(world!!, pos, side)?.also {
+					destinationCache[side.ordinal] = WeakReference(it)
+				}
+	}
+
+	private fun cableDisconnected(itf: SwitchInterface) {
+		macTable.entries.filter {
+			it.value == itf.side
+		}.forEach {
+			macTable.remove(it.key)
+		}
+		destinationCache[itf.side.ordinal] = null
+		flood(NetworkSplitFrame(itf.macAddress), itf)
 	}
 
 	fun tick() {
@@ -158,6 +173,10 @@ class SwitchBlockEntity(pos: BlockPos, state: BlockState): BlockEntity(PhyBlockE
 		override fun send(frame: EthernetFrame) {
 			switch.get()?.findDestination(this)?.receive(frame)
 		}
+
+		override fun cableDisconnected() {
+			switch.get()?.cableDisconnected(this)
+		}
 	}
 
 }

src/main/kotlin/net/shadowfacts/phycon/block/p2p/P2PReceiverBlockEntity.kt

@@ -12,9 +12,11 @@ import net.shadowfacts.phycon.api.packet.Packet
 import net.shadowfacts.phycon.api.util.IPAddress
 import net.shadowfacts.phycon.block.DeviceBlockEntity
 import net.shadowfacts.phycon.block.FaceDeviceBlock
+import net.shadowfacts.phycon.frame.NetworkSplitFrame
 import net.shadowfacts.phycon.init.PhyBlockEntities
 import net.shadowfacts.phycon.packet.*
 import net.shadowfacts.phycon.util.ClientConfigurableDevice
+import java.lang.ref.WeakReference
 
 /**
  * @author shadowfacts
@@ -36,6 +38,8 @@ class P2PReceiverBlockEntity(pos: BlockPos, state: BlockState): DeviceBlockEntit
     }
 
     companion object {
+        const val REQUEST_INVENTORY_TIMEOUT: Long = 100 // ticks
+
         fun provideItemStorage(be: P2PReceiverBlockEntity, side: Direction): Storage<ItemVariant>? {
             if (side == be.cachedState[FaceDeviceBlock.FACING]) {
                 return be.getTargetInventory()
@@ -46,21 +50,26 @@ class P2PReceiverBlockEntity(pos: BlockPos, state: BlockState): DeviceBlockEntit
 
     var target: IPAddress? = null
     var status = Status.NO_TARGET
+    private var requestTimestamp: Long = 0
 
     var clientObserver: (() -> Unit)? = null
 
     private var isFirstTick = true
-    // todo: need some way of removing this when there's no network path to the p2p interface
-    private var targetInventory: Storage<ItemVariant>? = null
+    private var targetInventory: WeakReference<Storage<ItemVariant>>? = null
 
     override fun handle(packet: Packet) {
         when (packet) {
             is PongPacket -> if (packet.source == target) status = Status.OK
-            is ReadItemStoragePacket -> targetInventory = packet.inventory
+            is ReadItemStoragePacket -> targetInventory = WeakReference(packet.inventory)
             is DeviceRemovedPacket -> if (packet.source == target) targetInventory = null
         }
     }
 
+    override fun handleNetworkSplit() {
+        super.handleNetworkSplit()
+        targetInventory = null
+    }
+
     override fun tick() {
         super.tick()
 
@@ -74,10 +83,12 @@ class P2PReceiverBlockEntity(pos: BlockPos, state: BlockState): DeviceBlockEntit
         if (target == null) {
             return null
         }
-        if (targetInventory == null) {
+        if (targetInventory?.get() == null && (counter - requestTimestamp) >= REQUEST_INVENTORY_TIMEOUT) {
+            status = Status.WAITING_FOR_RESPONSE
+            requestTimestamp = counter
             sendPacket(RequestInventoryPacket(RequestInventoryPacket.Kind.SIDED, ipAddress, target!!))
         }
-        return targetInventory
+        return targetInventory?.get()
     }
 
     private fun updateStatus() {

src/main/kotlin/net/shadowfacts/phycon/block/terminal/AbstractTerminalBlockEntity.kt

@@ -18,6 +18,7 @@ import net.shadowfacts.phycon.api.packet.Packet
 import net.shadowfacts.phycon.api.util.IPAddress
 import net.shadowfacts.phycon.block.DeviceBlockEntity
 import net.shadowfacts.phycon.component.*
+import net.shadowfacts.phycon.frame.NetworkSplitFrame
 import net.shadowfacts.phycon.packet.*
 import net.shadowfacts.phycon.util.NetworkUtil
 import java.lang.ref.WeakReference
@@ -38,7 +39,8 @@ abstract class AbstractTerminalBlockEntity(type: BlockEntityType<*>, pos: BlockP
 		// the locate/insertion timeouts are only 1 tick because that's long enough to hear from every device on the network
 		// in a degraded state (when there's latency in the network), not handling interface priorities correctly is acceptable
 		val LOCATE_REQUEST_TIMEOUT: Long = 1 // ticks
-		val INSERTION_TIMEOUT: Long = 1
+		val INSERTION_TIMEOUT: Long = 1 // ticks
+		val REQUEST_INVENTORY_TIMEOUT: Long = 1 // ticks
 	}
 
 	protected val inventoryCache = mutableMapOf<IPAddress, GroupedItemInvView>()
@@ -48,8 +50,8 @@ abstract class AbstractTerminalBlockEntity(type: BlockEntityType<*>, pos: BlockP
 	override val pendingInsertions = mutableListOf<PendingInsertion>()
 	override val dispatchStackTimeout = INSERTION_TIMEOUT
 
-	private var observers = 0
 	val cachedNetItems = ItemStackCollections.intMap()
+	private var requestInventoryTimestamp: Long? = null
 
 	// todo: multiple players could have the terminal open simultaneously
 	var netItemObserver: WeakReference<NetItemObserver>? = null
@@ -68,6 +70,11 @@ abstract class AbstractTerminalBlockEntity(type: BlockEntityType<*>, pos: BlockP
 		return null
 	}
 
+	override fun handleNetworkSplit() {
+		super.handleNetworkSplit()
+		inventoryCache.clear()
+	}
+
 	override fun handle(packet: Packet) {
 		when (packet) {
 			is ReadGroupedInventoryPacket -> handleReadInventory(packet)
@@ -164,14 +171,26 @@ abstract class AbstractTerminalBlockEntity(type: BlockEntityType<*>, pos: BlockP
 		if (!world!!.isClient) {
 			finishPendingRequests()
 			finishTimedOutPendingInsertions()
-		}
 
-		if (counter % 20 == 0L && !world!!.isClient) {
-			beginInsertions()
+			if (counter % 20 == 0L) {
+				beginInsertions()
+			}
+
+			if (requestInventoryTimestamp != null && (counter - requestInventoryTimestamp!!) >= REQUEST_INVENTORY_TIMEOUT) {
+				updateAndSync()
+				requestInventoryTimestamp = null
+			}
 		}
 	}
 
 	open fun onActivate(player: PlayerEntity) {
+		if (!world!!.isClient) {
+			updateAndSync()
+
+			inventoryCache.clear()
+			sendPacket(RequestInventoryPacket(RequestInventoryPacket.Kind.GROUPED, ipAddress))
+			requestInventoryTimestamp = counter
+		}
 	}
 
 	fun requestItem(stack: ItemStack, amount: Int = stack.count) {
@@ -212,7 +231,8 @@ abstract class AbstractTerminalBlockEntity(type: BlockEntityType<*>, pos: BlockP
 		// as with extracting, we "know" the new amounts and so can update instantly without actually sending out packets
 		updateAndSync()
 
-		return remaining
+		// don't start a second insertion, since remaining will be dispatched at the next beginInsertions
+		return ItemStack.EMPTY
 	}
 
 	override fun onInventoryChanged(inv: Inventory) {

src/main/kotlin/net/shadowfacts/phycon/block/terminal/CraftingTerminalBlockEntity.kt

@@ -33,11 +33,9 @@ class CraftingTerminalBlockEntity(pos: BlockPos, state: BlockState): AbstractTer
 	private val completedCraftingStackRequests = LinkedList<CraftingStackLocateRequest>()
 
 	override fun onActivate(player: PlayerEntity) {
-		if (!world!!.isClient) {
-			updateAndSync()
+		super.onActivate(player)
 
-			inventoryCache.clear()
-			sendPacket(RequestInventoryPacket(RequestInventoryPacket.Kind.GROUPED, ipAddress))
+		if (!world!!.isClient) {
 			val factory = object: ExtendedScreenHandlerFactory {
 				override fun createMenu(syncId: Int, playerInv: PlayerInventory, player: PlayerEntity): ScreenHandler? {
 					return CraftingTerminalScreenHandler(syncId, playerInv, this@CraftingTerminalBlockEntity)

src/main/kotlin/net/shadowfacts/phycon/block/terminal/TerminalBlockEntity.kt

@@ -18,11 +18,9 @@ import net.shadowfacts.phycon.packet.RequestInventoryPacket
 class TerminalBlockEntity(pos: BlockPos, state: BlockState): AbstractTerminalBlockEntity(PhyBlockEntities.TERMINAL, pos, state) {
 
 	override fun onActivate(player: PlayerEntity) {
-		if (!world!!.isClient) {
-			updateAndSync()
+		super.onActivate(player)
 
-			inventoryCache.clear()
-			sendPacket(RequestInventoryPacket(RequestInventoryPacket.Kind.GROUPED, ipAddress))
+		if (!world!!.isClient) {
 			val factory = object: ExtendedScreenHandlerFactory {
 				override fun createMenu(syncId: Int, playerInv: PlayerInventory, player: PlayerEntity): ScreenHandler {
 					return TerminalScreenHandler(syncId, playerInv, this@TerminalBlockEntity)

src/main/kotlin/net/shadowfacts/phycon/component/NetworkStackDispatcher.kt

@@ -36,7 +36,9 @@ interface NetworkStackDispatcher<Insertion: NetworkStackDispatcher.PendingInsert
 			insertion.results.add(packet.capacity to packet.stackReceiver)
 			if (insertion.isFinishable(this)) {
 				val remaining = finishInsertion(insertion)
-				// todo: do something with remaining
+				if (!remaining.isEmpty) {
+					pendingInsertions.add(createPendingInsertion(remaining))
+				}
 			}
 		}
 	}
@@ -56,12 +58,13 @@ interface NetworkStackDispatcher<Insertion: NetworkStackDispatcher.PendingInsert
 		// copy the insertion stack so subclasses that override this method can still see the originally dispatched stack after the super call
 		val remaining = insertion.stack.copy()
 		while (!remaining.isEmpty && sortedResults.isNotEmpty()) {
-			val (capacity, receivingInterface) = sortedResults.removeFirst()
+			val (capacity, receiver) = sortedResults.removeFirst()
 			if (capacity <= 0) continue
-			val copy = remaining.copyWithCount(min(capacity, remaining.count))
-			sendPacket(ItemStackPacket(copy, ipAddress, receivingInterface.ipAddress))
+			val sentCount = min(capacity, remaining.count)
+			val copy = remaining.copyWithCount(sentCount)
+			sendPacket(ItemStackPacket(copy, ipAddress, receiver.ipAddress))
 			// todo: the destination should confirm how much was actually inserted, in case of race condition
-			remaining.count -= capacity
+			remaining.count -= sentCount
 		}
 
 		return remaining
@@ -89,7 +92,11 @@ fun <Self, Insertion: NetworkStackDispatcher.PendingInsertion<Insertion>> Self.f
 
 	val finishable = pendingInsertions.filter { it.isFinishable(this) }
 	// finishInsertion removes the object from pendingInsertions
-	finishable.forEach(::finishInsertion)
-	// todo: do something with remaining?
+	for (insertion in finishable) {
+		val remaining = finishInsertion(insertion)
+		if (!remaining.isEmpty) {
+			pendingInsertions.add(createPendingInsertion(remaining))
+		}
+	}
 	// todo: if a timed-out insertion can't be finished, we should probably retry after some time (exponential backoff?)
 }

src/main/kotlin/net/shadowfacts/phycon/frame/ARPFrame.kt

@@ -1,5 +1,6 @@
 package net.shadowfacts.phycon.frame
 
+import net.shadowfacts.phycon.api.Interface
 import net.shadowfacts.phycon.api.util.IPAddress
 import net.shadowfacts.phycon.api.util.MACAddress
 

src/main/kotlin/net/shadowfacts/phycon/frame/NetworkSplitFrame.kt

@@ -0,0 +1,8 @@
+package net.shadowfacts.phycon.frame
+
+import net.shadowfacts.phycon.api.util.MACAddress
+
+/**
+ * @author shadowfacts
+ */
+class NetworkSplitFrame(source: MACAddress): BaseFrame(source, MACAddress.BROADCAST)