Skip to content

W
web
Commit fd50d6bd by James Yuan
Options

fix: 网关 Modbus 复用连接并加连接超时，避免单主机挂死拖垮其它网关 

mb-102 connect 阻塞会导致 mb-104 轮询超时被标成读失败；0.3.12 非阻塞 connect + Session 复用后 fail-fast。

Co-authored-by: Cursor <cursoragent@cursor.com>
parent 367d9363
Showing with 190 additions and 130 deletions
gateway/win-protocol/src/main.cpp
......@@ -113,7 +113,7 @@ struct AppState {
std::map<std::string, PointValue> values; // key = point id
std::map<std::string, bool> hostOk;
std::map<std::string, int> hostLatencyMs;
std::string version = "0.3.9-keep-good";
std::string version = "0.3.12-connect-timeout";
std::map<std::string, Session> sessions;
bool pollRunning = false;
};
......@@ -601,15 +601,27 @@ void pollOnce() {
int lastLat = 0;
int unit = mine.empty() ? 1 : mine.front().unitId;
std::map<int, uint16_t> flat;
// 同一主机整轮共用一条 TCP，避免每点新建连接把从站打满
modbus_tcp::Session sess;
const bool linked = sess.open(h.host, h.port);
if (!linked) {
hostOk[h.id] = false;
hostLatency[h.id] = 0;
std::cout << "[gateway] host " << h.id << " TCP connect failed " << h.host << ":" << h.port
<< "\n"
<< std::flush;
continue;
}
// 只采真实 Modbus；连不上 / 读失败不造模拟数
if (mine.empty()) {
std::vector<uint16_t> tmp;
int lat = 0;
const int probeLogical = 1;
const int probePdu = std::max(0, probeLogical + addrOffset);
bool ok = modbus_tcp::readHolding(h.host, h.port, unit, probePdu, 1, tmp, &lat);
bool ok = sess.readHolding(unit, probePdu, 1, tmp, &lat);
hostOk[h.id] = ok;
hostLatency[h.id] = lat;
sess.close();
continue;
}
// flat 按点表逻辑地址索引；读请求地址 = 逻辑地址 + offset
......@@ -626,7 +638,17 @@ void pollOnce() {
std::sort(addrs.begin(), addrs.end());
addrs.erase(std::unique(addrs.begin(), addrs.end()), addrs.end());
int chunksOk = 0, chunksFail = 0;
int singleOk = 0;
int consecutiveFail = 0;
constexpr int kMaxSingleFallback = 20; // 块失败后最多单点回退
constexpr int kAbortAfterFails = 3; // 连续失败则放弃本主机，避免拖死其它网关
for (size_t ai = 0; ai < addrs.size();) {
if (consecutiveFail >= kAbortAfterFails) {
std::cout << "[gateway] host " << h.id << " abort after " << consecutiveFail
<< " consecutive Modbus failures\n"
<< std::flush;
break;
}
int start = addrs[ai];
int end = start + 1;
size_t aj = ai + 1;
......@@ -640,33 +662,47 @@ void pollOnce() {
}
}
int cnt = std::min(chunk, std::max(1, end - start));
if (cnt > 125) cnt = 125;
const int pduStart = std::max(0, start + addrOffset);
std::vector<uint16_t> regs;
int lat = 0;
if (modbus_tcp::readHolding(h.host, h.port, unit, pduStart, cnt, regs, &lat) &&
if (sess.readHolding(unit, pduStart, cnt, regs, &lat, consecutiveFail == 0) &&
static_cast<int>(regs.size()) == cnt) {
anyOk = true;
lastLat = lat;
++chunksOk;
consecutiveFail = 0;
for (int i = 0; i < cnt; ++i) flat[start + i] = regs[static_cast<size_t>(i)];
} else {
++chunksFail;
for (size_t k = ai; k < aj; ++k) {
++consecutiveFail;
// 块读失败：有限次单点回退（复用同一连接）
bool gotOne = false;
for (size_t k = ai; k < aj && singleOk < kMaxSingleFallback; ++k) {
const int one = addrs[k];
const int onePdu = std::max(0, one + addrOffset);
std::vector<uint16_t> oneReg;
int oneLat = 0;
if (modbus_tcp::readHolding(h.host, h.port, unit, onePdu, 1, oneReg, &oneLat) &&
oneReg.size() == 1) {
if (sess.readHolding(unit, onePdu, 1, oneReg, &oneLat, false) && oneReg.size() == 1) {
anyOk = true;
lastLat = oneLat;
flat[one] = oneReg[0];
++chunksOk;
++singleOk;
gotOne = true;
consecutiveFail = 0;
}
}
if (!gotOne && consecutiveFail >= kAbortAfterFails) {
std::cout << "[gateway] host " << h.id << " abort after " << consecutiveFail
<< " consecutive Modbus failures\n"
<< std::flush;
break;
}
}
ai = aj > ai ? aj : ai + 1;
}
sess.close();
hostOk[h.id] = anyOk;
hostLatency[h.id] = lastLat;
std::cout << "[gateway] host " << h.id << " range=[" << minAddr << "," << maxEnd
......
gateway/win-protocol/src/modbus_tcp.hpp
#pragma once
/**
* 最小 Modbus TCP 客户端（FC03 Holding Register），Winsock。
* 最小 Modbus TCP 客户端（FC03 Holding / FC06 写寄存器），Winsock。
* 同一主机一轮采集内复用一条 TCP，避免每寄存器新建连接打满从站。
*/
#ifdef _WIN32
#ifndef WIN32_LEAN_AND_MEAN
......@@ -40,113 +41,6 @@ inline void wrBe16(uint8_t* p, uint16_t v) {
p[1] = static_cast<uint8_t>(v & 0xFF);
}
/** 读 Holding Registers；address 为 0-based PDU（调用方负责点表→PDU 偏移）。成功返回 true。 */
inline bool readHolding(const std::string& host, int port, int unitId, int address, int count,
std::vector<uint16_t>& out, int* latencyMs = nullptr) {
out.clear();
if (count <= 0 || count > 125) return false;
#ifdef _WIN32
if (!ensureWsa()) return false;
const auto t0 = GetTickCount64();
SOCKET sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (sock == INVALID_SOCKET) return false;
DWORD timeout = 800;
setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, reinterpret_cast<const char*>(&timeout), sizeof(timeout));
setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, reinterpret_cast<const char*>(&timeout), sizeof(timeout));
sockaddr_in addr{};
addr.sin_family = AF_INET;
addr.sin_port = htons(static_cast<u_short>(port));
if (InetPtonA(AF_INET, host.c_str(), &addr.sin_addr) != 1) {
addrinfo hints{};
hints.ai_family = AF_INET;
hints.ai_socktype = SOCK_STREAM;
addrinfo* res = nullptr;
if (getaddrinfo(host.c_str(), nullptr, &hints, &res) != 0 || !res) {
closesocket(sock);
return false;
}
addr.sin_addr = reinterpret_cast<sockaddr_in*>(res->ai_addr)->sin_addr;
freeaddrinfo(res);
}
if (connect(sock, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) != 0) {
closesocket(sock);
return false;
}
static uint16_t txn = 1;
uint16_t tid = txn++;
uint8_t req[12];
wrBe16(req + 0, tid);
wrBe16(req + 2, 0); // protocol
wrBe16(req + 4, 6); // length
req[6] = static_cast<uint8_t>(unitId & 0xFF);
req[7] = 0x03; // FC03
wrBe16(req + 8, static_cast<uint16_t>(address));
wrBe16(req + 10, static_cast<uint16_t>(count));
int sent = send(sock, reinterpret_cast<const char*>(req), 12, 0);
if (sent != 12) {
closesocket(sock);
return false;
}
uint8_t hdr[7];
int got = 0;
while (got < 7) {
int n = recv(sock, reinterpret_cast<char*>(hdr + got), 7 - got, 0);
if (n <= 0) {
closesocket(sock);
return false;
}
got += n;
}
uint16_t len = rdBe16(hdr + 4);
if (len < 3 || len > 260) {
closesocket(sock);
return false;
}
std::vector<uint8_t> body(len);
// unit already in hdr[6]; need remaining len-1 bytes after unit
body[0] = hdr[6];
int need = static_cast<int>(len) - 1;
got = 0;
while (got < need) {
int n = recv(sock, reinterpret_cast<char*>(body.data() + 1 + got), need - got, 0);
if (n <= 0) {
closesocket(sock);
return false;
}
got += n;
}
closesocket(sock);
if (latencyMs) *latencyMs = static_cast<int>(GetTickCount64() - t0);
// body: [unit][fc][bc][data...]
if (body.size() < 3) return false;
if (body[1] & 0x80) return false; // exception
if (body[1] != 0x03) return false;
uint8_t bc = body[2];
if (bc != count * 2 || body.size() < static_cast<size_t>(3 + bc)) return false;
out.resize(count);
for (int i = 0; i < count; ++i) {
out[i] = rdBe16(body.data() + 3 + i * 2);
}
return true;
#else
(void)host;
(void)port;
(void)unitId;
(void)address;
(void)latencyMs;
return false;
#endif
}
inline float regsToFloat(uint16_t hi, uint16_t lo) {
uint32_t bits = (uint32_t(hi) << 16) | uint32_t(lo);
float f = 0;
......@@ -189,13 +83,13 @@ inline bool txnExchange(SOCKET sock, const uint8_t* req, int reqLen, std::vector
return bodyOut.size() >= 2 && (bodyOut[1] & 0x80) == 0;
}
inline SOCKET connectTcp(const std::string& host, int port) {
inline SOCKET connectTcp(const std::string& host, int port, DWORD timeoutMs = 600) {
if (!ensureWsa()) return INVALID_SOCKET;
SOCKET sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (sock == INVALID_SOCKET) return INVALID_SOCKET;
DWORD timeout = 1200;
setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, reinterpret_cast<const char*>(&timeout), sizeof(timeout));
setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, reinterpret_cast<const char*>(&timeout), sizeof(timeout));
BOOL nodelay = TRUE;
setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, reinterpret_cast<const char*>(&nodelay),
sizeof(nodelay));
sockaddr_in addr{};
addr.sin_family = AF_INET;
addr.sin_port = htons(static_cast<u_short>(port));
......@@ -211,23 +105,112 @@ inline SOCKET connectTcp(const std::string& host, int port) {
addr.sin_addr = reinterpret_cast<sockaddr_in*>(res->ai_addr)->sin_addr;
freeaddrinfo(res);
}
if (connect(sock, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) != 0) {
// 非阻塞 connect + select，避免 SYN 阶段卡死整轮采集
u_long nonblock = 1;
ioctlsocket(sock, FIONBIO, &nonblock);
int cr = connect(sock, reinterpret_cast<sockaddr*>(&addr), sizeof(addr));
if (cr != 0) {
const int err = WSAGetLastError();
if (err != WSAEWOULDBLOCK && err != WSAEINPROGRESS) {
closesocket(sock);
return INVALID_SOCKET;
}
fd_set wset, eset;
FD_ZERO(&wset);
FD_ZERO(&eset);
FD_SET(sock, &wset);
FD_SET(sock, &eset);
timeval tv{};
tv.tv_sec = static_cast<long>(timeoutMs / 1000);
tv.tv_usec = static_cast<long>((timeoutMs % 1000) * 1000);
const int sel = select(0, nullptr, &wset, &eset, &tv);
if (sel <= 0 || FD_ISSET(sock, &eset)) {
closesocket(sock);
return INVALID_SOCKET;
}
int soerr = 0;
int solen = sizeof(soerr);
getsockopt(sock, SOL_SOCKET, SO_ERROR, reinterpret_cast<char*>(&soerr), &solen);
if (soerr != 0) {
closesocket(sock);
return INVALID_SOCKET;
}
}
nonblock = 0;
ioctlsocket(sock, FIONBIO, &nonblock);
setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, reinterpret_cast<const char*>(&timeoutMs),
sizeof(timeoutMs));
setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, reinterpret_cast<const char*>(&timeoutMs),
sizeof(timeoutMs));
return sock;
}
#endif
/** FC06 写单寄存器（现场统一用寄存器读写，禁用 FC02/FC05 线圈） */
inline bool writeSingleRegister(const std::string& host, int port, int unitId, int address,
uint16_t value) {
#ifdef _WIN32
SOCKET sock = connectTcp(host, port);
if (sock == INVALID_SOCKET) return false;
static uint16_t txn = 1;
/** 同一主机复用连接的会话（一轮 poll 内持有） */
struct Session {
SOCKET sock = INVALID_SOCKET;
std::string host;
int port = 0;
uint16_t txn = 1;
bool valid() const { return sock != INVALID_SOCKET; }
bool open(const std::string& h, int p) {
close();
host = h;
port = p;
sock = connectTcp(h, p);
return valid();
}
void close() {
if (sock != INVALID_SOCKET) {
closesocket(sock);
sock = INVALID_SOCKET;
}
}
bool reopen() { return open(host, port); }
bool readHolding(int unitId, int address, int count, std::vector<uint16_t>& out,
int* latencyMs = nullptr, bool allowReconnect = true) {
out.clear();
if (count <= 0 || count > 125) return false;
if (!valid() && !reopen()) return false;
const auto t0 = GetTickCount64();
uint8_t req[12];
wrBe16(req + 0, txn++);
if (txn == 0) txn = 1;
wrBe16(req + 2, 0);
wrBe16(req + 4, 6);
req[6] = static_cast<uint8_t>(unitId & 0xFF);
req[7] = 0x03;
wrBe16(req + 8, static_cast<uint16_t>(address));
wrBe16(req + 10, static_cast<uint16_t>(count));
std::vector<uint8_t> body;
if (!txnExchange(sock, req, 12, body)) {
// 仅允许一次重连，避免主机无响应时每点双倍超时拖死整轮
if (!allowReconnect || !reopen()) return false;
body.clear();
if (!txnExchange(sock, req, 12, body)) return false;
}
if (latencyMs) *latencyMs = static_cast<int>(GetTickCount64() - t0);
if (body.size() < 3 || body[1] != 0x03) return false;
uint8_t bc = body[2];
if (bc != count * 2 || body.size() < static_cast<size_t>(3 + bc)) return false;
out.resize(static_cast<size_t>(count));
for (int i = 0; i < count; ++i) {
out[static_cast<size_t>(i)] = rdBe16(body.data() + 3 + i * 2);
}
return true;
}
bool writeSingleRegister(int unitId, int address, uint16_t value) {
if (!valid() && !reopen()) return false;
uint8_t req[12];
wrBe16(req + 0, txn++);
if (txn == 0) txn = 1;
wrBe16(req + 2, 0);
wrBe16(req + 4, 6);
req[6] = static_cast<uint8_t>(unitId & 0xFF);
......@@ -235,11 +218,52 @@ inline bool writeSingleRegister(const std::string& host, int port, int unitId, i
wrBe16(req + 8, static_cast<uint16_t>(address));
wrBe16(req + 10, value);
std::vector<uint8_t> body;
bool ok = txnExchange(sock, req, 12, body) && body.size() >= 2 && body[1] == 0x06;
closesocket(sock);
if (!txnExchange(sock, req, 12, body)) {
if (!reopen()) return false;
body.clear();
if (!txnExchange(sock, req, 12, body)) return false;
}
return body.size() >= 2 && body[1] == 0x06;
}
};
#endif
/** 读 Holding（每次新建短连接；优先在 poll 中用 Session） */
inline bool readHolding(const std::string& host, int port, int unitId, int address, int count,
std::vector<uint16_t>& out, int* latencyMs = nullptr) {
#ifdef _WIN32
Session s;
if (!s.open(host, port)) return false;
bool ok = s.readHolding(unitId, address, count, out, latencyMs);
s.close();
return ok;
#else
(void)host; (void)port; (void)unitId; (void)address; (void)value;
(void)host;
(void)port;
(void)unitId;
(void)address;
(void)count;
(void)out;
(void)latencyMs;
return false;
#endif
}
/** FC06 写单寄存器 */
inline bool writeSingleRegister(const std::string& host, int port, int unitId, int address,
uint16_t value) {
#ifdef _WIN32
Session s;
if (!s.open(host, port)) return false;
bool ok = s.writeSingleRegister(unitId, address, value);
s.close();
return ok;
#else
(void)host;
(void)port;
(void)unitId;
(void)address;
(void)value;
return false;
#endif
}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment