diff --git a/base/hsocket.c b/base/hsocket.c index c6f023db1..1dcc8ae1c 100644 --- a/base/hsocket.c +++ b/base/hsocket.c @@ -67,6 +67,7 @@ int ResolveAddr(const char* host, sockaddr_u* addr) { if (inet_pton(AF_INET6, host, &addr->sin6.sin6_addr) == 1) { addr->sa.sa_family = AF_INET6; // host is ipv6 + return 0; } struct addrinfo* ais = NULL; @@ -190,11 +191,35 @@ int sockaddr_compare(const sockaddr_u* addr1, const sockaddr_u* addr2) { return addr1->sin6.sin6_family - addr2->sin6.sin6_family; if (addr1->sin6.sin6_port != addr2->sin6.sin6_port) return addr1->sin6.sin6_port - addr2->sin6.sin6_port; - return memcmp(&addr1->sin6.sin6_addr, &addr2->sin6.sin6_addr, sizeof(struct in_addr)); + return memcmp(&addr1->sin6.sin6_addr, &addr2->sin6.sin6_addr, sizeof(struct in6_addr)); } return memcmp(addr1, addr2, sizeof(sockaddr_u)); } +uint32_t sockaddr_ip_hash(sockaddr_u* addr) { + // FNV-1a hash over the ip bytes only (port excluded). + const uint32_t FNV_PRIME = 16777619u; + uint32_t hash = 2166136261u; + const unsigned char* p = NULL; + int len = 0; + if (addr->sa.sa_family == AF_INET) { + p = (const unsigned char*)&addr->sin.sin_addr; + len = sizeof(struct in_addr); + } + else if (addr->sa.sa_family == AF_INET6) { + p = (const unsigned char*)&addr->sin6.sin6_addr; + len = sizeof(struct in6_addr); + } + else { + return 0; + } + for (int i = 0; i < len; ++i) { + hash ^= p[i]; + hash *= FNV_PRIME; + } + return hash; +} + static int sockaddr_bind(sockaddr_u* localaddr, int type) { // socket -> setsockopt -> bind #ifdef SOCK_CLOEXEC diff --git a/base/hsocket.h b/base/hsocket.h index defd7d8ea..47169e3f9 100644 --- a/base/hsocket.h +++ b/base/hsocket.h @@ -122,6 +122,8 @@ HV_EXPORT int sockaddr_set_ipport(sockaddr_u* addr, const char* host, int port); HV_EXPORT socklen_t sockaddr_len(sockaddr_u* addr); HV_EXPORT const char* sockaddr_str(sockaddr_u* addr, char* buf, int len); HV_EXPORT int sockaddr_compare(const sockaddr_u* addr1, const sockaddr_u* addr2); +// hash the ip part only (port ignored), so the same client ip maps to the same value. used by LB_IpHash. +HV_EXPORT uint32_t sockaddr_ip_hash(sockaddr_u* addr); //#define INET_ADDRSTRLEN 16 //#define INET6_ADDRSTRLEN 46 diff --git a/docs/cn/TcpServer.md b/docs/cn/TcpServer.md index ec0932a15..e0ebdb06a 100644 --- a/docs/cn/TcpServer.md +++ b/docs/cn/TcpServer.md @@ -17,6 +17,8 @@ class TcpServer { void setMaxConnectionNum(uint32_t num); // 设置负载均衡策略 + // 可选: LB_RoundRobin(轮询, 默认) / LB_Random(随机) / + // LB_LeastConnections(最少连接数) / LB_IpHash(按客户端IP哈希, 同一IP固定分配到同一worker) void setLoadBalance(load_balance_e lb); // 设置线程数 diff --git a/evpp/EventLoopThreadPool.h b/evpp/EventLoopThreadPool.h index efdb3178d..3c501a36a 100644 --- a/evpp/EventLoopThreadPool.h +++ b/evpp/EventLoopThreadPool.h @@ -28,7 +28,8 @@ class EventLoopThreadPool : public Status { thread_num_ = num; } - EventLoopPtr nextLoop(load_balance_e lb = LB_RoundRobin) { + // @param hash: used by LB_IpHash (see sockaddr_ip_hash), ignored by other strategies. + EventLoopPtr nextLoop(load_balance_e lb = LB_RoundRobin, uint32_t hash = 0) { size_t numLoops = loop_threads_.size(); if (numLoops == 0) return NULL; size_t idx = 0; @@ -43,8 +44,12 @@ class EventLoopThreadPool : public Status { idx = i; } } + } else if (lb == LB_IpHash) { + idx = hash % numLoops; } else { - // Not Implemented + // Not Implemented, fallback to RoundRobin + if (++next_loop_idx_ >= numLoops) next_loop_idx_ = 0; + idx = next_loop_idx_ % numLoops; } return loop_threads_[idx]->loop(); } diff --git a/evpp/TcpServer.h b/evpp/TcpServer.h index 5d8c14403..ec1ebe9a3 100644 --- a/evpp/TcpServer.h +++ b/evpp/TcpServer.h @@ -255,7 +255,11 @@ class TcpServerEventLoopTmpl { TcpServerEventLoopTmpl* server = (TcpServerEventLoopTmpl*)hevent_userdata(connio); // NOTE: detach from acceptor loop hio_detach(connio); - EventLoopPtr worker_loop = server->worker_threads.nextLoop(server->load_balance); + uint32_t hash = 0; + if (server->load_balance == LB_IpHash) { + hash = sockaddr_ip_hash((sockaddr_u*)hio_peeraddr(connio)); + } + EventLoopPtr worker_loop = server->worker_threads.nextLoop(server->load_balance, hash); if (worker_loop == NULL) { worker_loop = server->acceptor_loop; }