bba: move frame queueing to net services. Call service to rx frame

New Service::sendEthFrame function to send frame to bba
Let net services implement queuing and synchronization
capcom: don't interfere when not in use

core/hw/bba/bba.cpp

@@ -23,48 +23,9 @@
 #include "serialize.h"
 #include "hw/sh4/sh4_sched.h"
 #include "cfg/option.h"
-#include <atomic>
-#include <deque>
-#include <mutex>
-#include <vector>
 
 static RTL8139State *rtl8139device;
 
-// The network backends (picotcp, DCNet) deliver frames from their own
-// threads, but the RTL8139 model and the SH-4 interrupt controller belong to
-// the emulation thread: raising the receive interrupt from another thread
-// races with SIIDRebuild() and can leave stray pending interrupts, which
-// the guest sees as an unhandled IRQ. Queue the frames here and hand them
-// to the chip from the SH-4 scheduler.
-static std::mutex rxMutex;
-static std::deque<std::vector<u8>> rxQueue;
-static std::atomic<bool> rxPending;
-static int rxSchedId = -1;
-constexpr size_t RX_QUEUE_MAX = 64;
-constexpr int RX_POLL_CYCLES = 1_sh4ms;
-
-static int rxDeliver(int tag, int cycles, int jitter, void *arg)
-{
-	if (rxPending.load(std::memory_order_acquire) && rtl8139device != nullptr)
-	{
-		std::lock_guard<std::mutex> _(rxMutex);
-		while (!rxQueue.empty() && rtl8139_can_receive(rtl8139device))
-		{
-			rtl8139_receive(rtl8139device, rxQueue.front().data(), rxQueue.front().size());
-			rxQueue.pop_front();
-		}
-		rxPending = !rxQueue.empty();
-	}
-	return RX_POLL_CYCLES;
-}
-
-static void rxClear()
-{
-	std::lock_guard<std::mutex> _(rxMutex);
-	rxQueue.clear();
-	rxPending = false;
-}
-
 // 1400 - 1600 GAPSPCI bridge registers
 // 1600 - 1700 standard PCI config
 // 1700 - 1800 rtl8139c I/O ports
@@ -91,6 +52,23 @@ static u8 GAPS_regs[GAPSPCI_REGS_SIZE];
 static u32 dmaOffset;
 static bool interruptPending;
 static constexpr u32 MAX_ETH_FRAME_SIZE = 1514;
+static int rxSchedId = -1;
+constexpr int RX_POLL_CYCLES = 5_sh4ms;
+
+static int rxDeliver(int tag, int cycles, int jitter, void *arg)
+{
+	u8 frame[MAX_ETH_FRAME_SIZE];
+	while (rtl8139_can_receive(rtl8139device))
+	{
+		const u32 size = net::modbba::sendEthFrame(frame);
+		if (size == 0)
+			break;
+		if (size < 6 || size > MAX_ETH_FRAME_SIZE)
+			continue;
+		rtl8139_receive(rtl8139device, frame, size);
+	}
+	return RX_POLL_CYCLES;
+}
 
 static void setInterrupt()
 {
@@ -116,7 +94,6 @@ static void bba_Init()
 	memcpy(&GAPS_regs[0], "GAPSPCI_BRIDGE_2", 0x10);
 	memcpy(&GAPS_regs[0x1c], "SEGA", 4);
 
-	rxClear();
 	rxSchedId = sh4_sched_register(0, rxDeliver);
 	sh4_sched_request(rxSchedId, RX_POLL_CYCLES);
 }
@@ -134,7 +111,6 @@ void bba_Term()
 		sh4_sched_unregister(rxSchedId);
 		rxSchedId = -1;
 	}
-	rxClear();
 }
 
 void bba_Reset(bool hard)
@@ -266,20 +242,6 @@ ssize_t qemu_send_packet(RTL8139State *s, const uint8_t *buf, int size)
 	return size;
 }
 
-// Called from the network threads
-int bba_recv_frame(const u8 *data, u32 len)
-{
-	if (len < 6 || len > MAX_ETH_FRAME_SIZE)
-		return 0;
-	std::lock_guard<std::mutex> _(rxMutex);
-	if (rxQueue.size() >= RX_QUEUE_MAX)
-		return 0;
-	rxQueue.emplace_back(data, data + len);
-	rxPending.store(true, std::memory_order_release);
-
-	return 1;
-}
-
 void pci_dma_read(PCIDevice *dev, dma_addr_t addr, void *buf, dma_addr_t len)
 {
 	memcpy(buf, &GAPS_ram[addr & GAPSPCI_RAM_MASK], len);

core/hw/bba/bba.h

@@ -25,4 +25,3 @@ u32 bba_ReadMem(u32 addr, u32 sz);
 void bba_WriteMem(u32 addr, u32 data, u32 sz);
 void bba_Serialize(Serializer& ser);
 void bba_Deserialize(Deserializer& deser);
-int bba_recv_frame(const u8 *data, u32 len);

core/network/capcom.cpp

@@ -301,9 +301,6 @@ public:
 		return rxQueue.size();
 	}
 
-	void receiveEthFrame(const u8 *frame, u32 size) override {
-	}
-
 	juice_agent_t *getJuiceAgent()
 	{
 		if (initStatus.valid())
@@ -775,13 +772,15 @@ void CapcomDirect::startCapcomDirectService()
 }
 void CapcomDirect::stopCapcomDirectService()
 {
-	setCustomService(nullptr);
-	if (capcomService != nullptr) {
+	if (active)
+		resetModemCallbacks();
+	if (capcomService != nullptr)
+	{
+		setCustomService(nullptr);
 		capcomService->realStop();
 		capcomService.reset();
+		INFO_LOG(NETWORK, "Capcom network service stopped");
 	}
-	resetModemCallbacks();
-	INFO_LOG(NETWORK, "Capcom network service stopped");
 }
 
 void CapcomDirect::onNetworkChange()
@@ -821,8 +820,8 @@ void CapcomDirect::onEvent(Event event, void *arg)
 		self->onLoadGame();
 		break;
 	case Event::Terminate:
-		self->active = false;
 		self->stopCapcomDirectService();
+		self->active = false;
 		break;
 	case Event::Network:
 		self->onNetworkChange();

core/network/dcnet.cpp

@@ -22,7 +22,6 @@
 #include "util/tsqueue.h"
 #include "oslib/oslib.h"
 #include "emulator.h"
-#include "hw/bba/bba.h"
 #include "cfg/option.h"
 #include "stdclass.h"
 #include "oslib/i18n.h"
@@ -55,6 +54,7 @@ public:
 	int modemAvailable() override;
 
 	void receiveEthFrame(const u8 *frame, u32 size) override;
+	u32 sendEthFrame(u8 *frame) override;
 };
 
 class PPPSocket
@@ -236,9 +236,9 @@ private:
 class EthSocket
 {
 public:
-	EthSocket(asio::io_context& io_context, const asio::ip::tcp::endpoint& endpoint,
+	EthSocket(asio::io_context& io_context, TsQueue<std::vector<u8>>& bbaQueue, const asio::ip::tcp::endpoint& endpoint,
 			const std::string& endpointName = "")
-		: socket(io_context)
+		: socket(io_context), bbaQueue(bbaQueue)
 	{
 		asio::error_code ec;
 		socket.connect(endpoint, ec);
@@ -319,7 +319,7 @@ private:
 					*(u16 *)&recvBuffer[14], (int)len - 2);
 				*/
 				ethdump(&recvBuffer[2], len - 2);
-				bba_recv_frame(&recvBuffer[2], len - 2);
+				bbaQueue.push(std::vector<u8>(&recvBuffer[2], &recvBuffer[len]));
 				if (len < recvBuffer.size())
 					recvBuffer.erase(recvBuffer.begin(), recvBuffer.begin() + len);
 				else
@@ -414,6 +414,7 @@ private:
 	std::array<u8, 1600> sendBuffer;
 	u32 sendBufferIdx = 0;
 	bool sending = false;
+	TsQueue<std::vector<u8>>& bbaQueue;
 
 	FILE *dumpfp = nullptr;
 };
@@ -662,6 +663,17 @@ public:
 		}
 	}
 
+	u32 recvEthFrame(u8 *frame)
+	{
+		if (bbaQueue.empty())
+			return 0;
+		const std::vector<u8> v = bbaQueue.pop();
+		if (v.size() > 1514)
+			return 0;
+		memcpy(frame, v.data(), v.size());
+		return v.size();
+	}
+
 private:
 	void run();
 	void connect(const asio::ip::address& address = {}, const std::string& apname = {});
@@ -670,6 +682,7 @@ private:
 	std::unique_ptr<asio::io_context> io_context;
 	std::unique_ptr<PPPSocket> pppSocket;
 	std::unique_ptr<EthSocket> ethSocket;
+	TsQueue<std::vector<u8>> bbaQueue;
 
 	static constexpr uint16_t PPP_PORT = 7654;
 	static constexpr uint16_t TAP_PORT = 7655;
@@ -737,6 +750,10 @@ void DCNetService::receiveEthFrame(u8 const *frame, unsigned int len)
 	thread.sendEthFrame(frame, len);
 }
 
+u32 DCNetService::sendEthFrame(u8 *frame) {
+	return thread.recvEthFrame(frame);
+}
+
 void DCNetThread::connect(const asio::ip::address& address, const std::string& apname)
 {
 	const bool powerSmash = settings.content.gameId == "HDR-0113"	// Power Smash
@@ -776,7 +793,7 @@ void DCNetThread::connect(const asio::ip::address& address, const std::string& a
 			endpoint.port(PPP_PORT);
 	}
 	if (config::EmulateBBA)
-		ethSocket = std::make_unique<EthSocket>(*io_context, endpoint, apname);
+		ethSocket = std::make_unique<EthSocket>(*io_context, bbaQueue, endpoint, apname);
 	else if (powerSmash)
 		pppSocket = std::make_unique<PowerSmashPPPSocket>(*io_context, endpoint, apname);
 	else
@@ -786,6 +803,7 @@ void DCNetThread::connect(const asio::ip::address& address, const std::string& a
 void DCNetThread::run()
 {
 	toModem.clear();
+	bbaQueue.clear();
 	try {
 		std::string hostname;
 #ifndef LIBRETRO

core/network/netservice.cpp

@@ -51,9 +51,13 @@ void stop()
 		service->stop();
 }
 
-void setCustomService(Service *service) {
+void setCustomService(Service *service)
+{
+	if (service == nullptr && net::modbba::service == customService)
+		net::modbba::service = nullptr;
+	else if (service != nullptr)
+		net::modbba::service = service;
 	customService = service;
-	net::modbba::service = service;
 }
 
 void writeModem(u8 b) {
@@ -69,9 +73,17 @@ int modemAvailable() {
 	return service->modemAvailable();
 }
 
-void receiveEthFrame(const u8 *frame, u32 size) {
-	start();
+void receiveEthFrame(const u8 *frame, u32 size)
+{
+	if (!settings.network.online)
+		start();
 	service->receiveEthFrame(frame, size);
 }
+u32 sendEthFrame(u8 *frame)
+{
+	if (!settings.network.online)
+		start();
+	return service->sendEthFrame(frame);
+}
 
 }

core/network/netservice.h

@@ -37,6 +37,7 @@ int readModem();
 int modemAvailable();
 
 void receiveEthFrame(const u8 *frame, u32 size);
+u32 sendEthFrame(u8 *frame);
 
 class Service
 {
@@ -49,7 +50,8 @@ public:
 	virtual int readModem() = 0;
 	virtual int modemAvailable() = 0;
 
-	virtual void receiveEthFrame(const u8 *frame, u32 size) = 0;
+	virtual void receiveEthFrame(const u8 *frame, u32 size) {}
+	virtual u32 sendEthFrame(u8 *frame) { return 0; }
 };
 
 void setCustomService(Service *service);

core/network/picoppp.cpp

@@ -51,7 +51,6 @@ extern "C" {
 #include "oslib/oslib.h"
 #include "util/tsqueue.h"
 #include "util/shared_this.h"
-#include "hw/bba/bba.h"
 
 #include <unordered_map>
 #include <deque>
@@ -69,6 +68,8 @@ static pico_device *pico_dev;
 static TsQueue<u8> in_buffer;
 static TsQueue<u8> out_buffer;
 
+static TsQueue<std::vector<u8>> bbaBuffer;
+
 static pico_ip4 dcaddr;
 static pico_ip4 dnsaddr;
 
@@ -282,8 +283,10 @@ public:
 		return socket;
 	}
 
-	void close() {
+	void close()
+	{
 		closeAll();
+		detachPicoSocket();
 		directPlay.reset();
 	}
 
@@ -1178,46 +1181,28 @@ static void closeDumpFile()
 		pcapngDump = nullptr;
 	}
 }
-// Frames sent by the emulated BBA arrive on the emulation thread. picotcp is
-// built without PICO_SUPPORT_MUTEX, so pico_queue_protect() does nothing and
-// calling pico_stack_recv() here corrupts the device queue that the pico
-// thread is draining. Queue the frames and feed them from the pico thread.
-static std::mutex ethInMutex;
-static std::deque<std::vector<u8>> ethInQueue;
-constexpr size_t ETH_IN_QUEUE_MAX = 256;
 
+// BBA -> pico stack
 static void pico_receive_eth_frame(const u8 *frame, u32 size)
 {
 	if (pico_dev == nullptr) {
 		start_pico();
 	}
 	else {
-		std::lock_guard<std::mutex> _(ethInMutex);
-		if (ethInQueue.size() < ETH_IN_QUEUE_MAX)
-			ethInQueue.emplace_back(frame, frame + size);
+		dumpFrame(frame, size);
+		pico_stack_recv(pico_dev, (u8 *)frame, size);
 	}
 }
 
-// Called on the pico thread
-static void feed_eth_frames()
+// pico stack -> BBA
+static int send_eth_frame(pico_device *dev, void *data, int len)
 {
-	std::deque<std::vector<u8>> frames;
-	{
-		std::lock_guard<std::mutex> _(ethInMutex);
-		frames.swap(ethInQueue);
-	}
-	if (pico_dev == nullptr)
-		return;
-	for (auto& frame : frames)
-	{
-		dumpFrame(frame.data(), frame.size());
-		pico_stack_recv(pico_dev, frame.data(), frame.size());
-	}
-}
-
-static int send_eth_frame(pico_device *dev, void *data, int len) {
 	dumpFrame((const u8 *)data, len);
-	return bba_recv_frame((const u8 *)data, len);
+	if (len < 6 || len > 1514)
+		// discard
+		return len;
+	bbaBuffer.push(std::vector<u8>((const u8 *)data, (const u8 *)data + len));
+	return len;
 }
 
 static void picoTick(const std::error_code& ec, asio::steady_timer *timer)
@@ -1226,7 +1211,6 @@ static void picoTick(const std::error_code& ec, asio::steady_timer *timer)
 		ERROR_LOG(NETWORK, "picoTick timer error: %s", ec.message().c_str());
 		return;
 	}
-	feed_eth_frames();
 	pico_stack_tick();
 	timer->expires_at(timer->expiry() + asio::chrono::milliseconds(PICO_TICK_MS));
 	timer->async_wait(std::bind(picoTick, asio::placeholders::error, timer));
@@ -1312,6 +1296,7 @@ void PicoThread::run()
 	// Empty queues
 	in_buffer.clear();
 	out_buffer.clear();
+	bbaBuffer.clear();
 
     // Find DNS ip address
 	{
@@ -1488,8 +1473,6 @@ static void stop_pico()
 	emu.setNetworkState(false);
 	pico_thread_running = false;
 	pico_thread.stop();
-	std::lock_guard<std::mutex> _(ethInMutex);
-	ethInQueue.clear();
 }
 
 // picotcp mutex implementation
@@ -1537,5 +1520,14 @@ void PicoTcpService::receiveEthFrame(const u8 *frame, u32 size) {
 	pico_receive_eth_frame(frame, size);
 }
 
+u32 PicoTcpService::sendEthFrame(u8 *frame)
+{
+	if (bbaBuffer.empty())
+		return 0;
+	const std::vector<u8> v = bbaBuffer.pop();
+	memcpy(frame, v.data(), v.size());
+	return v.size();
+}
+
 }
 

core/network/picoppp.h

@@ -33,6 +33,7 @@ public:
 	int readModem() override;
 	int modemAvailable() override;
 	void receiveEthFrame(const u8 *frame, u32 size) override;
+	u32 sendEthFrame(u8 *frame) override;
 };
 
 }

core/network/rawmodem.h

@@ -36,8 +36,6 @@ public:
 	int readModem() override;
 	int modemAvailable() override;
 
-	void receiveEthFrame(const u8 *frame, u32 size) override {};
-
 private:
 	uint16_t port;
 };

core/util/tsqueue.h

@@ -20,6 +20,7 @@
 #include <queue>
 #include <mutex>
 #include <condition_variable>
+#include <utility>
 
 template<typename T>
 class TsQueue