SequenceLogic/Sleds
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
b56ef122dd
Sleds/K2Daemon/ClientCache.cpp

1521 lines
48 KiB
C++
Raw Normal View History

Initial commit part 2
2025-03-13 21:28:38 +00:00
/*
Copyright (c) 2013 IOnU Security Inc. All rights reserved
Created August 2013 by Kendrick Webster
K2Daemon/ClientCache.cpp - implementation for ClientCache.h
*/
#if !defined(NDEBUG)
#define BOOST_MULTI_INDEX_ENABLE_INVARIANT_CHECKING
#define BOOST_MULTI_INDEX_ENABLE_SAFE_MODE
#endif
#include <boost/multi_index_container.hpp>
#include <boost/multi_index/hashed_index.hpp>
#include <boost/multi_index/member.hpp>
#include <boost/multi_index/ordered_index.hpp>
#include <iostream>
#include <string>
#include <chrono>
#include <list>
#include <unordered_map>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <stdint.h>
#include <string.h>
#include "global.h"
#include "Version.h"
#include "MongoNotifyQueue.h"
#include "MongoDeviceStatus.h"
#include "OfficeSubscriptions.h"
#include "ClientMap.h"
#include "ClientCache.h"
using boost::multi_index_container;
using namespace boost::multi_index;
/* --------------------------------------------------------------------------
Data structures representing connected clients and their states
Most of the time, clients will simply heartbeat while not being involved
in message delivery attempts. Such clients are efficiently represented by
this data structure.
This data structure is kept as small as possible. Transient data is kept
in separate parallel data structures.
*/
struct CClient
{
ClientMap::id_t client_id;
std::string fragment_accumulator; // to receive messages spanning multiple packets (to do: move to parallel datastruct)
uint32_t ip4_addr;
uint16_t udp_port;
uint8_t tx_sequence;
uint8_t rx_sequence;
uint32_t client_instance;
uint16_t protocol_version;
uint64_t login_time;
uint64_t time; // (future) time of next action if no packet is received from client
};
// A boost::multi_index_container facilitates efficient access of CClient nodes
// based on a hash map of IDs or an ordered set of timestamps.
struct by_id {};
struct by_time {};
typedef multi_index_container<
CClient,
indexed_by<
hashed_unique<
tag<by_id>,
BOOST_MULTI_INDEX_MEMBER(CClient, ClientMap::id_t, client_id)
>,
ordered_non_unique<
tag<by_time>,
BOOST_MULTI_INDEX_MEMBER(CClient, uint64_t, time)
>
>
> client_cache_t;
typedef client_cache_t::index<by_time>::type& client_time_index_t;
typedef client_cache_t::index<by_id>::type& client_id_index_t;
typedef client_cache_t::index<by_time>::type::iterator client_time_iterator_t;
typedef client_cache_t::index<by_id>::type::iterator client_id_iterator_t;
/* --------------------------------------------------------------------------
Data structures representing message delivery attempts and retry timers
When a message is being sent to a client, it will be cached in this data
structure to facilitate retries and timeouts. Records are purged from
this data structure when clients return to an idle state.
*/
struct CRetry
{
ClientMap::id_t client_id;
std::string nqueue_id; // MongoDB notification queue document ID (for K2IPC_MESSAGE packets)
std::string info; // packet content (body)
uint16_t state; // packet retry state
uint16_t opcode; // packet type
uint64_t time; // (future) time of next action if no packet is received from client
};
enum // packet retry states
{
STATE_PACKET_RETRY1,
STATE_PACKET_RETRY2,
STATE_PACKET_RETRY3,
STATE_PACKET_TIMEOUT
};
typedef multi_index_container<
CRetry,
indexed_by<
hashed_unique<
tag<by_id>,
BOOST_MULTI_INDEX_MEMBER(CRetry, ClientMap::id_t, client_id)
>,
ordered_non_unique<
tag<by_time>,
BOOST_MULTI_INDEX_MEMBER(CRetry, uint64_t, time)
>
>
> retry_cache_t;
typedef retry_cache_t::index<by_id>::type& retry_id_index_t;
typedef retry_cache_t::index<by_time>::type& retry_time_index_t;
typedef retry_cache_t::index<by_id>::type::iterator retry_id_iterator_t;
typedef retry_cache_t::index<by_time>::type::iterator retry_time_iterator_t;
/* --------------------------------------------------------------------------
Data structures for message queueing
When a packet needs to be sent to a client that is already active in a
ping-pong ACK transaction, the packet will be queued.
*/
struct CDeferredPacket
{
uint16_t opcode; // packet type
std::string info; // packet content (body)
CDeferredPacket(uint16_t op, const char* inf)
: opcode(op), info(inf) {}
};
typedef std::list<CDeferredPacket> defer_queue_t;
typedef std::unordered_map<ClientMap::id_t, defer_queue_t> defer_map_t;
static bool send_deferred;
/* --------------------------------------------------------------------------
Local (module static) data and constants
*/
// Timing constants
#define RETRY1_MS (K2DMN_RETRY1_TENTHS_OF_SECONDS * 100)
#define RETRY2_MS ((K2DMN_RETRY2_TENTHS_OF_SECONDS - K2DMN_RETRY1_TENTHS_OF_SECONDS) * 100)
#define RETRY3_MS ((K2DMN_RETRY3_TENTHS_OF_SECONDS - K2DMN_RETRY2_TENTHS_OF_SECONDS) * 100)
#define TIMEOUT_MS ((K2DMN_RETRY3_TIMEOUT_TENTHS_OF_SECONDS - K2DMN_RETRY3_TENTHS_OF_SECONDS) * 100)
#define HEARTBEAT_TIMEOUT_MS (K2CLI_HEARTBEAT_TIMEOUT_SECONDS * 1000)
// Major data structures
static client_cache_t client_cache;
static retry_cache_t retry_cache;
static defer_map_t defer_map;
// Received packet, parsed
namespace RxPacket
{
// Fields common to all packet types
uint32_t client_instance;
uint16_t protocol_version;
uint8_t sequence;
uint8_t opcode;
bool repeated;
const char* body;
const char* client_urn;
ClientMap::id_t client_id;
// K2IPC_SUBSCRIBE_OFFICE
struct
{
char fragmentation;
std::string client_urn;
std::string offices;
}
subscribe_office;
}
// Packet encoding/decoding constants
namespace PacketEncoding
{
static const uint8_t key [K2IPC_KEY_SIZE] = {K2IPC_KEY};
static const uint8_t mic [K2IPC_MIC_SIZE] = {K2IPC_MIC};
}
namespace ClientStats
{
using Memstats::uint_t;
uint_t logins;
uint_t logouts;
uint_t timeouts;
// for calculating average login time
uint_t login_time_sum;
uint_t login_time_count;
// for including currently-connected clients in average login time
uint_t connected_timestamp_sum;
uint_t connected_timestamp_count;
}
/* --------------------------------------------------------------------------
Low-level support functions
*/
// returns (future) timestamp in microseconds with added offset,
// t0 (timestamp value 0) is set by the first call to this function
static uint64_t timestamp(unsigned int milliseconds)
{
using std::chrono::duration_cast;
using std::chrono::microseconds;
using std::chrono::steady_clock;
steady_clock::time_point now = steady_clock::now();
static steady_clock::time_point start = now;
uint64_t elapsed_microseconds = duration_cast<microseconds>(now - start).count();
return elapsed_microseconds + (1000 * milliseconds);
}
static const char* fragmentDescriptorToString(char fd)
{
switch (fd)
{
case K2IPC_FRAGMENT_FIRST: return "FIRST";
case K2IPC_FRAGMENT_SINGLE: return "SINGLE";
case K2IPC_FRAGMENT_MIDDLE: return "MIDDLE";
case K2IPC_FRAGMENT_LAST: return "LAST";
default: return "<ERROR>";
}
}
static std::string ip4_string(struct in_addr s)
{
return std::string(inet_ntoa(s));
}
static std::string ip4_string(uint32_t a)
{
struct in_addr s;
s.s_addr = a;
return ip4_string(s);
}
static const char* opcodeName(unsigned int n)
{
switch (n)
{
#define X(op) case K2IPC_##op: return #op;
K2IPC_PACKET_OPCODES
#undef X
}
return "<BAD_OPCODE>";
}
static void trackLoginTime(const CClient& client)
{
ClientStats::login_time_sum += (timestamp(0) - client.login_time);
ClientStats::login_time_count++;
ClientStats::connected_timestamp_sum -= client.login_time;
ClientStats::connected_timestamp_count--;
}
static double getSecondsOnlineAverage(void)
{
using ClientStats::uint_t;
uint_t sum = ClientStats::login_time_sum;
uint_t n = ClientStats::login_time_count;
// include currently-connected clients
uint_t c = ClientStats::connected_timestamp_count;
if (c)
{
double d = ClientStats::connected_timestamp_sum;
d /= c;
d = timestamp(0) - d;
return ((c * d + sum) / 1E6) / (c + n);
}
else
{
return n ? ((sum / 1E6) / n) : 0.0;
}
}
/* --------------------------------------------------------------------------
Client state support functions
*/
// fetch a const pointer to a client
static const CClient* getClient(ClientMap::id_t client_id)
{
client_id_index_t idx = client_cache.get<by_id>();
client_id_iterator_t it = idx.find(client_id);
if (it == idx.end())
{
return NULL;
}
else
{
return &*it;
}
}
// remove client from client cache
static void removeFromClientCache(ClientMap::id_t client_id)
{
client_id_index_t idx = client_cache.get<by_id>();
client_id_iterator_t it = idx.find(client_id);
if (it != idx.end())
{
trackLoginTime(*it);
idx.erase(it);
logprintf(Log::debug2, "removed client id(%d) urn(%s) from client cache", client_id, ClientMap::UrnFromId(client_id));
}
else
{
logprintf(Log::error, "removeFromClientCache: client id(%d) urn(%s) not found", client_id, ClientMap::UrnFromId(client_id));
}
}
// remove client from defer queue
static void removeFromDeferQueue(ClientMap::id_t client_id)
{
defer_map_t::iterator it = defer_map.find(client_id);
if (defer_map.end() != it)
{
defer_map.erase(it);
}
}
// remove client from retry cache and defer queue
static void removeFromRetryCache(const char* client_urn)
{
ClientMap::id_t client_id = ClientMap::IdFromUrn(client_urn);
retry_id_index_t idx = retry_cache.get<by_id>();
retry_id_iterator_t it = idx.find(client_id);
if (it != idx.end())
{
idx.erase(it);
logprintf(Log::debug2, "removed client id(%d) urn(%s) from retry cache", client_id, client_urn);
}
// else {expected}, usually client won't be found
removeFromDeferQueue(client_id);
}
// client went away (logged out, timed out, etc)
// WARNING: since this unmaps the client, it should be done AFTER anything that uses map
static void clientOffline(const char* client_urn)
{
OfficeSubscriptions::Subscribe(client_urn, "");
MongoDB::nqueue::UpdateAllDeviceRecords(client_urn, MongoDB::nqueue::TYPE_PUSH);
ClientMap::Unmap(client_urn);
}
/* --------------------------------------------------------------------------
Packet encoding and sending
*/
/* get/update the sequence number for a transmitted packet */
static uint8_t getTxSequence(CClient& client, uint8_t opcode, bool retry)
{
switch (opcode)
{
case K2IPC_MESSAGE:
case K2IPC_DEVICE_STATUS:
case K2IPC_PING:
if (!retry)
{
++client.tx_sequence;
}
// logprintf(Log::debug3, "sending %s with seq %d", opcodeName(opcode), client.tx_sequence);
return client.tx_sequence;
case K2IPC_ACK_CONNECT:
case K2IPC_NAK_CONNECT:
case K2IPC_ACK_LOGOUT:
case K2IPC_ACK_POLL_DB:
case K2IPC_ACK_HEARTBEAT:
case K2IPC_ACK_SUBSCRIBE_OFFICE:
case K2IPC_ACK_QUERY_STATS:
// logprintf(Log::debug3, "sending %s with seq %d", opcodeName(opcode), client.rx_sequence);
return client.rx_sequence;
default:
logprintf(Log::error, "unsupported opcode (%d, %s) for getTxSequence", opcode, opcodeName(opcode));
return 0;
}
}
/*
Encode a packet
Returns length of packet in octets, 0 if error
*/
static unsigned int encodePacket(CClient& client, uint8_t opcode, const char* info, bool retry)
{
using PacketEncoding::key;
using PacketEncoding::mic;
sc_hash_state_t h;
uint8_t* p = tx_buf;
int len = strlen(info);
/* sanity-check string length */
if ((K2IPC_OVERHEAD_SIZE + len) > K2IPC_MAX_PACKET_SIZE)
{
logprintf(Log::error, "string length (%d) too large for packet (%d max)",
len, K2IPC_MAX_PACKET_SIZE - K2IPC_OVERHEAD_SIZE);
return 0;
}
/* generate nonce and use it to key the stream cipher */
memset(&h, 0, sizeof(h));
sc_get_vector(&rng_hash, p, K2IPC_NONCE_SIZE);
sc_hash_vector(&h, p, K2IPC_NONCE_SIZE);
sc_hash_vector(&h, key, sizeof(key));
sc_hash_mix(&h, K2IPC_KEY_MIX_CYCLES);
/* add other packet fields, encrypt packet, return length */
p += K2IPC_NONCE_SIZE;
*p++ = ((client.protocol_version >> 8) & 0xFF);
*p++ = (client.protocol_version & 0xFF);
*p++ = ((client.client_instance >> 16) & 0xFF);
*p++ = ((client.client_instance >> 8) & 0xFF);
*p++ = (client.client_instance & 0xFF);
*p++ = getTxSequence(client, opcode, retry);
*p++ = opcode;
memcpy(p, info, len + 1);
memcpy(p + len + 1, mic, K2IPC_MIC_SIZE);
sc_hash_encrypt(&h, tx_buf + K2IPC_NONCE_SIZE, len + K2IPC_CRYPT_SIZE);
return (unsigned int)len + K2IPC_OVERHEAD_SIZE;
}
static void sendPacket(CClient& client, uint8_t opcode, const char* info, bool retry = false)
{
unsigned int len = encodePacket(client, opcode, info, retry);
if (0 < len)
{
struct sockaddr_in addr;
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = client.ip4_addr;
addr.sin_port = client.udp_port;
sendto(sockfd, tx_buf, len, 0, reinterpret_cast<struct sockaddr *>(&addr), sizeof(addr));
}
}
// Sends a packet that the client should ACK, updates the retry cache and/or defer queue
static ClientCache::send_message_result_t sendMessage
(
CClient& client,
uint8_t opcode,
const char* info,
const char* nqueue_id = NULL, // only used for MESSAGE packets, for fetching database record
bool use_defer_queue = true // in-memory queue is used to defer packets other than MESSAGE
)
{
retry_id_index_t ridx = retry_cache.get<by_id>();
retry_id_iterator_t rit = ridx.find(client.client_id);
if (rit == ridx.end())
{
CRetry r;
r.client_id = client.client_id;
r.opcode = opcode;
r.info = info;
if (nqueue_id)
{
r.nqueue_id = nqueue_id;
}
r.state = STATE_PACKET_RETRY1;
r.time = timestamp(RETRY1_MS);
ridx.insert(r);
sendPacket(client, opcode, info);
return ClientCache::SENT_PENDING_ACK;
}
else
{
if (use_defer_queue)
{
defer_map[client.client_id].emplace_back(opcode, info);
logprintf(Log::debug, "Deferring %s for %s:%d id(%d) urn(%s)",
opcodeName(opcode), ip4_string(client.ip4_addr).c_str(), ntohs(client.udp_port),
client.client_id, ClientMap::UrnFromId(client.client_id));
}
return ClientCache::CLIENT_BUSY;
}
}
// calls the main overload of sendMessage(...) with <client> fetched from client_cache
static ClientCache::send_message_result_t sendMessage
(
ClientMap::id_t client_id,
uint8_t opcode,
const char* info,
const char* nqueue_id = NULL,
bool use_defer_queue = true,
bool was_deferred = false // enables log output for showing what happens to deferred packets
)
{
client_id_index_t idx = client_cache.get<by_id>();
client_id_iterator_t it = idx.find(client_id);
if (it == idx.end())
{
return ClientCache::CLIENT_OFFLINE;
}
CClient c = *it;
if (was_deferred)
{
logprintf(Log::debug, "Sending deferred %s for %s:%d id(%d) urn(%s)",
opcodeName(opcode), ip4_string(c.ip4_addr).c_str(), ntohs(c.udp_port),
client_id, ClientMap::UrnFromId(client_id));
}
ClientCache::send_message_result_t r = sendMessage(c, opcode, info, nqueue_id, use_defer_queue);
idx.replace(it, c);
return r;
}
// checks the defer queue for a client, sends the next queued message if present
static void sendDeferred(ClientMap::id_t client_id)
{
defer_map_t::iterator it = defer_map.find(client_id);
if (defer_map.end() != it)
{
defer_queue_t& queue = it->second;
if (queue.empty())
{
logprintf(Log::error, "sendDeferred: did not expect an empty queue to be mapped");
defer_map.erase(it);
return;
}
CDeferredPacket& p = queue.front();
switch (sendMessage(client_id, p.opcode, p.info.c_str(), NULL, true, true))
{
case ClientCache::SENT_PENDING_ACK:
// expected, do nothing
break;
case ClientCache::CLIENT_BUSY: // not expected, this function should only be called when the client is not busy
logprintf(Log::error, "sendDeferred: client id(%d) BUSY", client_id);
break;
case ClientCache::CLIENT_OFFLINE: // not expected, an offline device should have no retry cache or defer queue activity
logprintf(Log::error, "sendDeferred: client id(%d) OFFLINE", client_id);
break;
default: // added a return code?
logprintf(Log::error, "sendDeferred: unexpected return code from sendMessage() for client id(%d)", client_id);
break;
}
queue.pop_front();
if (queue.empty())
{
defer_map.erase(it);
}
}
}
/* --------------------------------------------------------------------------
Packet decoding and parsing
*/
/*
Decode a packet
Returns pointer to sequence number on success, NULL if packet is bad
*/
static const uint8_t* decodePacket(unsigned int len)
{
using PacketEncoding::key;
using PacketEncoding::mic;
sc_hash_state_t h;
uint8_t* p = rx_buf;
/* reject short packets */
if (len < K2IPC_OVERHEAD_SIZE)
{
logprintf(Log::debug3, "spurious (short) packet from %s:%d", ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port));
return NULL;
}
/* decrypt packet */
memset(&h, 0, sizeof(h));
sc_hash_vector(&h, p, K2IPC_NONCE_SIZE);
sc_hash_vector(&h, key, sizeof(key));
sc_hash_mix(&h, K2IPC_KEY_MIX_CYCLES);
p += K2IPC_NONCE_SIZE;
sc_hash_decrypt(&h, p, len - K2IPC_NONCE_SIZE);
/* reject packet if MIC is wrong */
if (0 != memcmp(rx_buf + (len - K2IPC_MIC_SIZE), mic, K2IPC_MIC_SIZE))
{
logprintf(Log::debug3, "spurious (bad MIC) packet from %s:%d", ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port));
return NULL;
}
/* reject packet if NUL terminator for body (string) is missing */
if (0 != rx_buf[len - (K2IPC_MIC_SIZE + 1)])
{
logprintf(Log::warning, "malformed (missing NUL) packet from %s:%d", ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port));
return NULL;
}
return p;
}
// helper for parsePacketBody --
// parse K2IPC_SUBSCRIBE_OFFICE
static success_t parseSubscribeOffice(void)
{
using RxPacket::body;
using RxPacket::client_urn;
using RxPacket::subscribe_office;
const char* p = strchr(body, K2IPC_URN_DELIMITER);
if (p)
{
size_t len = p - body;
subscribe_office.client_urn.assign(body, len);
client_urn = subscribe_office.client_urn.c_str();
subscribe_office.fragmentation = *++p;
subscribe_office.offices = ++p;
return SUCCESS;
}
else
{
logprintf(Log::error, "parseSubscribeOffice: missing delimiter, body = \"%s\"", body);
return FAILURE;
}
}
// helper for parsePacket --
// parse <RxPacket::body> into subfields based on opcode
static success_t parsePacketBody(void)
{
using RxPacket::opcode;
using RxPacket::body;
using RxPacket::client_urn;
switch (opcode)
{
case K2IPC_CONNECT:
case K2IPC_HEARTBEAT:
case K2IPC_LOGOUT:
case K2IPC_POLL_DB:
case K2IPC_QUERY_STATS:
case K2IPC_ACK_MESSAGE:
case K2IPC_ACK_DEVICE_STATUS:
case K2IPC_ACK_PING:
client_urn = body;
return SUCCESS;
case K2IPC_SUBSCRIBE_OFFICE:
return parseSubscribeOffice();
default:
logprintf(Log::error, "parsePacketBody: unsupported opcode(%d, %s)", opcode, opcodeName(opcode));
return FAILURE;
}
}
// Parse a received packet
// Returns SUCCESS if the packet is good
static success_t parsePacket(unsigned int len)
{
using RxPacket::client_instance;
using RxPacket::protocol_version;
using RxPacket::sequence;
using RxPacket::opcode;
using RxPacket::body;
const uint8_t* p = decodePacket(len);
if (NULL == p)
{
return FAILURE;
}
protocol_version = *p++;
protocol_version <<= 8;
protocol_version |= *p++;
client_instance = *p++;
client_instance <<= 8;
client_instance |= *p++;
client_instance <<= 8;
client_instance |= *p++;
sequence = *p++;
opcode = *p++;
body = reinterpret_cast<const char *>(p);
return parsePacketBody();
}
/* --------------------------------------------------------------------------
Handlers for received message (non-ACK) packets
*/
// handle a K2IPC_CONNECT packet
static void onConnect(CClient& client)
{
using RxPacket::client_urn;
sendPacket(client, K2IPC_ACK_CONNECT, client_urn);
}
// handle a K2IPC_HEARTBEAT packet
static void onHeartbeat(CClient& client)
{
using RxPacket::client_urn;
sendPacket(client, K2IPC_ACK_HEARTBEAT, client_urn);
}
// handle a K2IPC_LOGOUT packet
static void onLogout(CClient& client)
{
using RxPacket::client_urn;
sendPacket(client, K2IPC_ACK_LOGOUT, client_urn);
}
// garbage collect K2IPC_LOGOUT
static void onLogoutGC(void)
{
using RxPacket::client_id;
using RxPacket::client_urn;
if (!RxPacket::repeated)
{
logprintf(Log::debug2, "LOGOUT from %s:%d id(%d) urn(%s)",
ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port), client_id, client_urn);
OfficeSubscriptions::NotifyStatus(client_urn, OfficeSubscriptions::OFFLINE);
MongoDB::devstatus::Update(client_urn, MongoDB::devstatus::STATUS_OFFLINE, ip4_string(recvaddr.sin_addr.s_addr).c_str());
removeFromClientCache(client_id);
removeFromRetryCache(client_urn);
clientOffline(client_urn);
ClientStats::logouts++;
}
else
{
logprintf(Log::error, "repeated onLogoutGC unexpected, client id(%d) urn(%s) should have been unmapped",
client_id, client_urn);
}
}
// handle a K2IPC_POLL_DB packet
static void onPollDatabase(CClient& client)
{
sendPacket(client, K2IPC_ACK_POLL_DB, RxPacket::client_urn);
if (!RxPacket::repeated)
{
logprintf(Log::debug3, "POLL_DB from %s:%d urn(%s)",
ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port), RxPacket::client_urn);
Main::PollDatabase();
}
}
// handle a K2IPC_SUBSCRIBE_OFFICE packet
static void onSubscribeOffice(CClient& client)
{
using RxPacket::client_urn;
using RxPacket::subscribe_office;
sendPacket(client, K2IPC_ACK_SUBSCRIBE_OFFICE, client_urn);
if (!RxPacket::repeated)
{
if (subscribe_office.offices.length() > 40)
{
logprintf(Log::debug3, "%s SUBSCRIBE_OFFICE frag %-7s = %s ...",
client_urn, fragmentDescriptorToString(subscribe_office.fragmentation), subscribe_office.offices.substr(0, 40).c_str());
}
else
{
logprintf(Log::debug3, "%s SUBSCRIBE_OFFICE frag %-7s = %s",
client_urn, fragmentDescriptorToString(subscribe_office.fragmentation), subscribe_office.offices.c_str());
}
switch (subscribe_office.fragmentation)
{
case K2IPC_FRAGMENT_FIRST:
case K2IPC_FRAGMENT_SINGLE:
client.fragment_accumulator = subscribe_office.offices;
break;
case K2IPC_FRAGMENT_MIDDLE:
case K2IPC_FRAGMENT_LAST:
client.fragment_accumulator += subscribe_office.offices;
break;
}
switch (subscribe_office.fragmentation)
{
case K2IPC_FRAGMENT_SINGLE:
case K2IPC_FRAGMENT_LAST:
OfficeSubscriptions::Subscribe(client_urn, client.fragment_accumulator.c_str());
client.fragment_accumulator.clear();
client.fragment_accumulator.shrink_to_fit();
break;
}
}
}
// handle a K2IPC_QUERY_STATS packet
static void onQueryStats(CClient& client)
{
using RxPacket::client_urn;
char buf[K2IPC_MAX_STRING_LENGTH];
int len = snprintf(buf, sizeof(buf), "\
{\n\
\"system_stats\": [\n\
{\n\
\"val\": %lu,\n\
\"key\": \"CpuCount\"\n\
},\n\
{\n\
\"val\": %lu,\n\
\"key\": \"TotalVirtualMemory\"\n\
},\n\
{\n\
\"val\": %lu,\n\
\"key\": \"TotalMemory\"\n\
},\n\
{\n\
\"val\": %lu,\n\
\"key\": \"VirtualMemoryUsed\"\n\
},\n\
{\n\
\"val\": %lu,\n\
\"key\": \"MemoryUsed\"\n\
},\n\
{\n\
\"val\": %0.1lf,\n\
\"key\": \"CpuLoadPercentOfAllCores\"\n\
}\n\
],\n\
\"daemon_stats\": [\n\
{\n\
\"val\": %lu,\n\
\"key\": \"UptimeSeconds\"\n\
},\n\
{\n\
\"val\": %lu,\n\
\"key\": \"Logins\"\n\
},\n\
{\n\
\"val\": %lu,\n\
\"key\": \"Logouts\"\n\
},\n\
{\n\
\"val\": %lu,\n\
\"key\": \"Timeouts\"\n\
},\n\
{\n\
\"val\": %0.0lf,\n\
\"key\": \"SecondsOnlineAvg\"\n\
},\n\
{\n\
\"val\": %lu,\n\
\"key\": \"ConnectedClients\"\n\
},\n\
{\n\
\"val\": %lu,\n\
\"key\": \"SubscribedOffices\"\n\
},\n\
{\n\
\"val\": %lu,\n\
\"key\": \"OfficeSubscriptions\"\n\
},\n\
{\n\
\"val\": %lu,\n\
\"key\": \"VirtualMemoryUsed\"\n\
},\n\
{\n\
\"val\": %lu,\n\
\"key\": \"MemoryUsed\"\n\
},\n\
{\n\
\"val\": %lu,\n\
\"key\": \"HeapAllocated\"\n\
},\n\
{\n\
\"val\": %0.1lf,\n\
\"key\": \"CpuLoadPercentOfOneCore\"\n\
}\n\
],\n\
\"class\": \"%s\",\n\
\"version\": \"%u.%u.%u\",\n\
\"revised\": \"%s\"\n\
}\n",
Memstats::CpuCount(),
Memstats::TotalVirtualMemory(),
Memstats::TotalMemory(),
Memstats::SystemWideVirtualMemoryUsed(),
Memstats::SystemWideMemoryUsed(),
Memstats::SystemWideCpuLoadPercent(),
Memstats::UptimeSeconds(),
ClientStats::logins,
ClientStats::logouts,
ClientStats::timeouts,
getSecondsOnlineAverage(),
Memstats::ConnectedClients(),
Memstats::SubscribedOffices(),
Memstats::OfficeSubscriptions(),
Memstats::VirtualMemoryUsed(),
Memstats::MemoryUsed(),
Memstats::HeapAllocated(),
Memstats::CpuLoadPercent(),
Version::Data::name,
Version::Data::major,
Version::Data::minor,
Version::Data::step,
Version::Data::timestamp
);
sendPacket(client, K2IPC_ACK_QUERY_STATS, buf);
logprintf(Log::debug3, "QUERY_STATS from %s:%d urn(%s), result length %d",
ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port), client_urn, len);
if (len >= static_cast<int>(sizeof(buf)))
{
logprintf(Log::error, "memory stats query result truncated from %d bytes to %d", len, sizeof(buf) - 1);
}
}
/* --------------------------------------------------------------------------
Handlers for received ACK packets
*/
// generic ACK handler with callback for type-specific handling of original (non-repeated) ACK
static void onAck(void (* onOriginalAck)(const CRetry&))
{
if (!RxPacket::repeated)
{
uint8_t tx_opcode_expected;
switch (RxPacket::opcode)
{
case K2IPC_ACK_MESSAGE: tx_opcode_expected = K2IPC_MESSAGE; break;
case K2IPC_ACK_DEVICE_STATUS: tx_opcode_expected = K2IPC_DEVICE_STATUS; break;
case K2IPC_ACK_PING: tx_opcode_expected = K2IPC_PING; break;
default:
logprintf(Log::error, "onAck: unhandled opcode(%d, %s)", RxPacket::opcode, opcodeName(RxPacket::opcode));
return;
}
retry_id_index_t idx = retry_cache.get<by_id>();
retry_id_iterator_t it = idx.find(RxPacket::client_id);
if (it == idx.end())
{
logprintf(Log::debug, "onAck: empty retry cache (repeated ACK) for %s from %s:%d urn(%s)", opcodeName(RxPacket::opcode),
ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port), RxPacket::client_urn);
}
else if (tx_opcode_expected != it->opcode)
{
logprintf(Log::error, "onAck: received %s from %s:%d urn(%s), sent %s (not %s)", opcodeName(RxPacket::opcode),
ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port), RxPacket::client_urn,
opcodeName(it->opcode), opcodeName(tx_opcode_expected));
}
else
{
if (onOriginalAck)
{
onOriginalAck(*it);
}
idx.erase(it);
send_deferred = true;
}
}
}
// handle a K2IPC_ACK_MESSAGE packet
static void onAckMessage_(const CRetry& r)
{
MongoDB::nqueue::Delete(r.nqueue_id.c_str());
logprintf(Log::debug3, "ACK_MESSAGE from %s:%d urn(%s)",
ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port), RxPacket::client_urn);
MongoDB::devstatus::Update(RxPacket::client_urn, MongoDB::devstatus::STATUS_NOTIFIED, ip4_string(recvaddr.sin_addr.s_addr).c_str());
}
static void onAckMessage(void)
{
onAck(onAckMessage_);
}
// handle a K2IPC_ACK_DEVICE_STATUS packet
static void onAckDeviceStatus(void)
{
onAck(NULL);
}
// handle a K2IPC_ACK_PING packet
static void onAckPing_(const CRetry& r)
{
logprintf(Log::informational, "received ACK_PING from %s:%d urn(%s)",
ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port), RxPacket::client_urn);
}
static void onAckPing(void)
{
onAck(onAckPing_);
}
/* --------------------------------------------------------------------------
Plumbing for received packets
*/
// invoke handler for a parsed <RxPacket>
static void dispatchPacket(CClient& client)
{
using RxPacket::opcode;
switch (opcode)
{
case K2IPC_CONNECT: onConnect(client); break;
case K2IPC_HEARTBEAT: onHeartbeat(client); break;
case K2IPC_LOGOUT: onLogout(client); break;
case K2IPC_POLL_DB: onPollDatabase(client); break;
case K2IPC_SUBSCRIBE_OFFICE: onSubscribeOffice(client); break;
case K2IPC_QUERY_STATS: onQueryStats(client); break;
case K2IPC_ACK_MESSAGE: onAckMessage(); break;
case K2IPC_ACK_DEVICE_STATUS: onAckDeviceStatus(); break;
case K2IPC_ACK_PING: onAckPing(); break;
default:
logprintf(Log::error, "dispatchPacket: unsupported opcode (%d, %s)", opcode, opcodeName(opcode));
break;
}
}
// invoke garbage collector (2nd-half handler that can modify client cache) for a parsed <RxPacket>
static void garbageCollectPacket(void)
{
switch (RxPacket::opcode)
{
case K2IPC_LOGOUT:
onLogoutGC();
break;
}
}
// helper for handleParsedPacket --
// check/update the sequence number in <client> for a received packet in <RxPacket>
enum
{
RX_SEQUENCE_GOOD,
RX_SEQUENCE_REPEATED,
RX_SEQUENCE_BAD
};
static int checkRxSequence(CClient& client)
{
using RxPacket::sequence;
using RxPacket::opcode;
using RxPacket::client_urn;
switch (opcode)
{
case K2IPC_CONNECT:
client.tx_sequence = client.rx_sequence = 0;
logprintf(Log::debug, "repeated CONNECT from %s:%d urn(%s)",
ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port), client_urn);
return RX_SEQUENCE_REPEATED;
case K2IPC_HEARTBEAT:
case K2IPC_LOGOUT:
case K2IPC_POLL_DB:
case K2IPC_SUBSCRIBE_OFFICE:
case K2IPC_QUERY_STATS:
if (++client.rx_sequence == sequence)
{
return RX_SEQUENCE_GOOD;
}
else
{
if (--client.rx_sequence == sequence)
{
repeated_packet:
logprintf(Log::debug, "repeated packet with opcode(%d, %s) from %s:%d urn(%s)",
opcode, opcodeName(opcode), ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port), client_urn);
return RX_SEQUENCE_REPEATED;
}
else
{
logprintf(Log::notice, "expected sequence %d, received sequence %d for packet with opcode(%d, %s) from %s:%d urn(%s)",
(client.rx_sequence + 1) & 0xFF, sequence, opcode, opcodeName(opcode),
ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port), client_urn);
}
}
return RX_SEQUENCE_BAD;
case K2IPC_ACK_MESSAGE:
case K2IPC_ACK_DEVICE_STATUS:
case K2IPC_ACK_PING:
if (client.tx_sequence == sequence)
{
return RX_SEQUENCE_GOOD;
}
else
{
if ((client.tx_sequence - 1) == sequence)
{
goto repeated_packet;
}
else
{
logprintf(Log::notice, "expected ACK sequence %d, received sequence %d for packet with opcode(%d, %s) from %s:%d urn(%s)",
client.tx_sequence, sequence, opcode, opcodeName(opcode),
ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port), client_urn);
}
}
return RX_SEQUENCE_BAD;
default:
logprintf(Log::error, "checkRxSequence: unsupported packet opcode (%d, %s) from %s:%d urn(%s)",
opcode, opcodeName(opcode), ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port), client_urn);
return RX_SEQUENCE_BAD;
}
}
// helper for handleParsedPacket - add a new client
// handles/dispatches K2IPC_CONNECT
static void addClient(void)
{
using RxPacket::client_urn;
CClient client;
client.client_id = ClientMap::Map(client_urn);
client.ip4_addr = recvaddr.sin_addr.s_addr;
client.udp_port = recvaddr.sin_port;
client.protocol_version = RxPacket::protocol_version;
client.client_instance = RxPacket::client_instance;
ClientStats::connected_timestamp_sum += client.login_time = timestamp(0);
ClientStats::connected_timestamp_count++;
client.time = timestamp(HEARTBEAT_TIMEOUT_MS);
client.tx_sequence = client.rx_sequence = 0;
logprintf(Log::debug2, "New client: id(%d), urn(%s), ip4_addr(%s), udp_port(%d), instance(%d), protocol_version(%d)",
client.client_id, client_urn, ip4_string(client.ip4_addr).c_str(), ntohs(client.udp_port),
client.client_instance, client.protocol_version);
MongoDB::devstatus::Update(client_urn, MongoDB::devstatus::STATUS_ONLINE, ip4_string(client.ip4_addr).c_str());
OfficeSubscriptions::NotifyStatus(client_urn, OfficeSubscriptions::ONLINE);
ClientStats::logins++;
dispatchPacket(client);
client_cache.get<by_id>().insert(client);
}
// helper for handleParsedPacket -
// log a warning for a packet received from an unconnected client,
// but only if its not the same client repeatedly hammering the server
static void logUnconnectedDevice(void)
{
using RxPacket::opcode;
using RxPacket::client_urn;
static std::string last_client;
if (last_client != client_urn)
{
last_client = client_urn;
logprintf(Log::warning, "received packet(s) with opcode(%d, %s) from unconnected client %s:%d urn(%s)",
opcode, opcodeName(opcode), ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port), client_urn);
}
}
// helper for handleParsedPacket -
// check the source IP/port for a received packet
// returns <true> if the address or port changed
static bool isSourceAddressChanged(const CClient& client)
{
return ((client.ip4_addr != recvaddr.sin_addr.s_addr) || (client.udp_port != recvaddr.sin_port));
}
// helper for handleConnectAddressChanged -
// send a NAK_CONNECT to the device from which <RxPacket> arrived
static void sendNakConnect(void)
{
CClient c;
c.ip4_addr = recvaddr.sin_addr.s_addr;
c.udp_port = recvaddr.sin_port;
c.client_instance = RxPacket::client_instance;
c.protocol_version = RxPacket::protocol_version;
c.tx_sequence = c.rx_sequence = 0;
sendPacket(c, K2IPC_NAK_CONNECT, RxPacket::client_urn);
}
// helper for handleParsedPacket -
// handle K2IPC_CONNECT with URN matching connected client but different src addr
static void handleConnectAddressChanged(ClientMap::id_t client_id, CClient& client)
{
logprintf(Log::informational, "sending NAK_CONNECT to %s:%d, sending PING to %s:%d",
ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port),
ip4_string(client.ip4_addr).c_str(), ntohs(client.udp_port));
sendNakConnect();
sendMessage(client, K2IPC_PING, RxPacket::client_urn);
}
// checks/updates client sequence numbers and timers for a received parsed packet,
// calls dispatchPacket(client) and returns SUCCESS if all is well
static success_t handleParsedPacket(void)
{
using RxPacket::opcode;
using RxPacket::client_urn;
ClientMap::id_t client_id = ClientMap::IdFromUrn(client_urn);
if (ClientMap::NULL_ID == client_id)
{
switch (opcode)
{
case K2IPC_CONNECT:
addClient();
break;
case K2IPC_LOGOUT:
logprintf(Log::informational, "LOGOUT ignored from %s:%d urn(%s), client is not logged in",
ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port), client_urn);
return FAILURE;
default:
logUnconnectedDevice();
return FAILURE;
}
}
else
{
RxPacket::client_id = client_id;
client_id_index_t idx = client_cache.get<by_id>();
client_id_iterator_t it = idx.find(client_id);
if (it == idx.end())
{
logprintf(Log::error, "id(%d) not found in client cache for urn(%s), attempting to handle packet from %s:%d",
client_id, client_urn, ip4_string(recvaddr.sin_addr.s_addr).c_str(), ntohs(recvaddr.sin_port));
return FAILURE;
}
else
{
CClient client = *it;
if (isSourceAddressChanged(client))
{
switch (opcode)
{
case K2IPC_CONNECT:
logprintf(Log::informational, "CONNECT from %s:%d, expected %s:%d for client id(%d) urn(%s)",
ip4_string(recvaddr.sin_addr).c_str(), ntohs(recvaddr.sin_port),
ip4_string(client.ip4_addr).c_str(), ntohs(client.udp_port),
client_id, client_urn);
handleConnectAddressChanged(client_id, client);
idx.replace(it, client);
return FAILURE;
case K2IPC_LOGOUT:
logprintf(Log::informational, "LOGOUT ignored from %s:%d, expected %s:%d for client id(%d) urn(%s)",
ip4_string(recvaddr.sin_addr).c_str(), ntohs(recvaddr.sin_port),
ip4_string(client.ip4_addr).c_str(), ntohs(client.udp_port),
client_id, client_urn);
return FAILURE;
default:
logprintf(Log::informational, "Client address changed: %s:%d --> %s:%d, client id(%d) urn(%s), packet opcode(%d, %s)",
ip4_string(client.ip4_addr).c_str(), ntohs(client.udp_port),
ip4_string(recvaddr.sin_addr).c_str(), ntohs(recvaddr.sin_port),
client_id, client_urn, opcode, opcodeName(opcode));
client.ip4_addr = recvaddr.sin_addr.s_addr;
client.udp_port = recvaddr.sin_port;
break;
}
}
RxPacket::repeated = false;
switch (checkRxSequence(client))
{
case RX_SEQUENCE_REPEATED:
RxPacket::repeated = true;
// FALL THROUGH
case RX_SEQUENCE_GOOD:
dispatchPacket(client);
client.time = timestamp(HEARTBEAT_TIMEOUT_MS);
idx.replace(it, client);
break;
case RX_SEQUENCE_BAD:
return FAILURE;
}
}
}
if (send_deferred)
{
send_deferred = false;
// must be called when no iterator on client_cache is held
sendDeferred(RxPacket::client_id);
}
return SUCCESS;
}
/* --------------------------------------------------------------------------
Timer-driven logic
*/
// timer subfunction: check for heartbeat timeouts
static void timerCheckClients(uint64_t now)
{
client_time_index_t idx = client_cache.get<by_time>();
client_time_iterator_t it = idx.begin();
client_time_iterator_t end = idx.end();
while (it != end)
{
if (it->time > now)
{
break;
}
else
{
const char* client_urn = ClientMap::UrnFromId(it->client_id);
if (client_urn)
{
logprintf(Log::debug2, "heartbeat timeout: client id(%d) urn(%s)", it->client_id, client_urn);
MongoDB::devstatus::Update(client_urn, MongoDB::devstatus::STATUS_TIMEOUT, "");
OfficeSubscriptions::NotifyStatus(client_urn, OfficeSubscriptions::OFFLINE);
removeFromRetryCache(client_urn);
clientOffline(client_urn);
}
else
{
logprintf(Log::error, "heartbeat timeout for unmapped client with id(%d)", it->client_id);
}
client_time_iterator_t i2 = it++;
trackLoginTime(*i2);
idx.erase(i2);
ClientStats::timeouts++;
}
}
}
// client did not ACK a packet within retry/timeout limits
static void onPacketTimeout(const CRetry& r)
{
const char* client_urn = ClientMap::UrnFromId(r.client_id);
if (client_urn)
{
const CClient* c = getClient(r.client_id);
std::string ipAddy;
if (c)
{
logprintf(Log::informational, "message timeout: client id(%d) urn(%s) addr(%s:%d) opcode(%d, %s)",
r.client_id, client_urn, ip4_string(c->ip4_addr).c_str(), ntohs(c->udp_port), r.opcode, opcodeName(r.opcode));
ipAddy = ip4_string(c->ip4_addr);
}
else
{
logprintf(Log::error, "unable to fetch CClient for message timeout: client id(%d) urn(%s) opcode(%d, %s)",
r.client_id, client_urn, r.opcode, opcodeName(r.opcode));
}
OfficeSubscriptions::NotifyStatus(client_urn, OfficeSubscriptions::OFFLINE);
removeFromDeferQueue(r.client_id);
removeFromClientCache(r.client_id);
MongoDB::devstatus::Update(client_urn, MongoDB::devstatus::STATUS_NO_ACK, ipAddy.c_str());
clientOffline(client_urn);
ClientStats::timeouts++;
}
else
{
logprintf(Log::error, "message timeout for client id(%d) not expected, client was already unmapped", r.client_id);
}
}
// retry sending a packet
static void retrySend(const CRetry& r)
{
client_id_index_t idx = client_cache.get<by_id>();
client_id_iterator_t it = idx.find(r.client_id);
if (it == idx.end())
{
logprintf(Log::error, "retrySend: client id(%d) urn(%s) not found", r.client_id, ClientMap::UrnFromId(r.client_id));
}
else
{
CClient c = *it;
sendPacket(c, r.opcode, r.info.c_str(), true);
idx.replace(it, c);
}
}
// timer subfunction: check for retries
static void timerCheckRetries(uint64_t now)
{
retry_time_index_t idx = retry_cache.get<by_time>();
retry_time_iterator_t it = idx.begin(), end = idx.end(), i2;
while (it != end)
{
if (it->time > now)
{
break;
}
CRetry r = *it;
switch (it->state)
{
case STATE_PACKET_RETRY1:
r.state = STATE_PACKET_RETRY2;
r.time = now + (1000 * RETRY2_MS);
send_again:
idx.replace(it, r);
retrySend(r);
++it;
break;
case STATE_PACKET_RETRY2:
r.state = STATE_PACKET_RETRY3;
r.time = now + (1000 * RETRY3_MS);
goto send_again;
case STATE_PACKET_RETRY3:
r.state = STATE_PACKET_TIMEOUT;
r.time = now + (1000 * TIMEOUT_MS);
goto send_again;
case STATE_PACKET_TIMEOUT:
onPacketTimeout(r);
i2 = it++;
idx.erase(i2);
break;
}
}
}
/* --------------------------------------------------------------------------
Module interface
*/
void ClientCache::HandlePacket(unsigned int len)
{
if (SUCCESS == parsePacket(len))
{
if (SUCCESS == handleParsedPacket())
{
garbageCollectPacket();
}
}
}
ClientCache::send_message_result_t ClientCache::SendMessage(const char* nqueue_id, const char* client_urn, const char* info)
{
return sendMessage(ClientMap::IdFromUrn(client_urn), K2IPC_MESSAGE, info, nqueue_id, false);
}
void ClientCache::SendDeviceStatus(ClientMap::id_t client_id, const char* info)
{
switch (sendMessage(client_id, K2IPC_DEVICE_STATUS, info))
{
case CLIENT_BUSY:
case SENT_PENDING_ACK:
// expected, do nothing
break;
case CLIENT_OFFLINE: // not expected, no reason to try to send device status to an offline device
logprintf(Log::error, "SendDeviceStatus: client id(%d) OFFLINE", client_id);
break;
default: // added a return code?
logprintf(Log::error, "SendDeviceStatus: unexpected return code from sendMessage() for client id(%d)", client_id);
break;
}
}
void ClientCache::Timer()
{
uint64_t now = timestamp(0);
timerCheckClients(now);
timerCheckRetries(now);
#if 0 // enable for testing as needed, disable for releases
static bool once = true;
if (once)
{
once = false;
logprintf(Log::debug, "sizeof(CClient) = %d", sizeof(CClient));
}
#endif
}
/*
Supports the --dump-clients <n> command-line option. The entire client list
is dumped to the log.
WARNING: will cause daemon unresponsiveness if the number of clients is
large and/or the log sink is slow.
*/
void ClientCache::DumpClients()
{
client_time_index_t idx = client_cache.get<by_time>();
client_time_iterator_t it = idx.begin();
client_time_iterator_t end = idx.end();
logprintf(Log::informational, "<------- Connected Clients ------------");
while (it != end)
{
logprintf(Log::informational, "%s", ClientMap::UrnFromId(it->client_id));
++it;
}
logprintf(Log::informational, "-------- End Connected Clients ------->");
}
Reference in New Issue Copy Permalink