aioenetd.cpp

// aioenetd.cpp
// Main source file for aioenetd, the systemd service/application
//  that listens to some TCP ports and provides the packet-level interface to eNET- devices

/*
	when a connection to ~8080 (the "control" connection port) occurs send a Hello 'H' TMessage with several TDataItems.
		At least one data item is the ConnectionID, others may include PID, Model#, Serial#, what the onboard RTC thinks the time is, etc
		DId 0x7001 "TCP_ConnectionID", 4 byte ConnectionID is the data
	When a connection to ~8080+1 (the ADC Streaming port) occurs, send (__u32)(0x80000000|ConnectionID) ("Invalid ADC data value bit set + ConnectionID").

	Connection IDs are opaque daemon-issued tokens, not Linux socket descriptors.
	ADC_StreamStart(ConnectionID) resolves the ADC token to a still-connected data socket.
*/

/*[aioenetd Protocol 2 TCP-Listener/Server Daemon/Service implementation and concept notes]
from discord code-review conversation with Daria; these do not belong in this source file:
	your "the main loop" == "my worker thread that dequeues Actions";
	your "exploding" == "my Object factory construction";
	You've moved "my Object construction" into a single-threaded spot, "the main loop", from where I have it, in an
	individual socket's receive-thread.
	Because I have multiple receive threads it isn't nearly as necessary to "be fast": the TCP Stack will queue bytes for me.
	Because my error-checking location (the parser, the exploder, .FromBytes()) is in a socket-specific (ie client-specific)
	thread it is harde\
	{			  \
		;		  \
	}
	Because my worker and all receive threads share a thread-safe std::queue<> everything is serialized nicely.

	By putting 66% of the error checking, syntax AND semantic (but not operational errors, eg hardware timeouts) into the
	receive threads — and in fact, into the TMessage constructor, I am guaranteed all TMessage Objects are valid and safe
	to submit to the worker thread, thus less likely to cause errors in that single-thread / single-point-of-failure
	(and make the worker execution faster, as it is "my single thread" and thus bottleneck)
-----
	A TMessage is constructed from received bytes by `auto aMessage = TMessage::fromBytes(buf);`, or an "X" Response TMessage with
	syntax error details gets returned, instead.
	Either way, the constructed TMessage is pushed into the Action Queue.
	The asynchronous worker thread pops a TMessage off the Action Queue, does a `for(aDataItem : aMessage.Payload){aDataItem.Go()}`,
	and either modifies-in-place and sends the TMessage as a reply or constructs and sends a new TMessage as the reply,
	the TMessage(s) then goes out of scope and deallocates
-----
	The TMessage library needs to handle syntax errors, mistakes in the *format* of a bytestream, and semantic errors, mistakes
	in the *content* of the received bytes.  Some categoriese of semantic errors, however, are specific to a particular model of eNET-
	device, let alone specific to a model Family.

	Consider ADC_GetChannelV(iChannel): iChannel is valid if (0 <= iChannel <= 15), right?  Nope, not "generally": this is only true
	for the ~12 models in the base Family, eNET-AIO16-16F, eNET-AI12-16E, etc.  But eNET- devices, and therefore Protocol 2 devices
	intended to operate via this TMessage library and the aioenetd implementation, include the DPK and DPK M Families; this means:
	iChannel is valid if (0 <= iChannel <= highestAdcChan), and highestAdcChan is 15, 31, 63, 96, or 127; depending on the specific
	model running aioenetd/using this library.

	So, to catch Semantic errors of this type (invalid channel parameter in an ADC_GetChannelV() TDataItem) the parser must "know"
	the value of "highestAdcChan" for the model it is running on.

	The sum of all things that the parser needs to know to handle semantic error checking, specific to the model running the library,
	are encapsulated as "getters" in a HAL.  The list is quite long as there are a LOT of variations built out of the eNET-AIO design.

	These "getters" *could* be implemented in a static, compile-time, manner.  Consider a device_specific.h file that has a bunch of
	eg `#define highestAdcChan 31`-type constants defined.  However, this is a "write it for 1" approach, and would require loading a
	different TMessage library binary for every model, as the binary is effectively hard-coded for a specific device's needs.

	The USB FWE2.0 firmwares are almost this simplistic in their approach to a HAL: there is a device_specific.h, but there is also a
	run-time operation that introspects the DeviceID and tweaks some constants, like the friendly_model_name string, to a model-specific
	value.  This run-time operation is a hard-coded switch(DeviceId){}, and thus is implemented per Firmware (but allows one Firmware
	to run, as a binary, on any models built from the same or compatible PCB).

	This only works because there are very few variants per firmware source: the USB designs are sufficiently different that reusing
	firmware is implausible: it would require a *real* HAL implementation (one that abstracted every pin on the FX2, and every
	register that could ever exist on the external address/data bus).

	We're going for a better approach to the HAL in this library.

	TBD, LOL

	The library implementation, today (2022-07-21 @ 10:55am Pacific), only checks for Semantic errors in the one DId that's been
	implemented; REG_Read1(offset).  This DId uses a helper function hardcoded in the source called `WidthFromOffset(offset)`,
	which returns one of 0, 8, or 32, indicating "invalid offset", "offset is a valid 8-bit register", or "offset is a valid
	32-bit register", respectively.  In theory it should be able to respond "16" as well, but the eNET-AIO, in all of its models,
	has no 16-bit registers.  This is an instance where the HAL is weak; "WidthFromOffset()" shouldn't be hard-coded; it should
	be able to respond accurately about whatever device it is running on (or perhaps even whatever device it is asked about).

	One way to accomplish this would be loading configuration information tables from nonvolatile on-device storage.  "tables",
	here, is plural not to refer to both a hypothetical single table needed for WidthFromOffset() and all the other tables for
	supporting other HAL-queriables, but to express that WidthFromOffset() *alone* needs several tables, if it is to support
	the general case.  Sure, eNET-AIO only has registers that can each only be correctly accessed at a specific bit width (ie it is
	unsafe or impossible to successfully read or write 8 bits from any eNET-AIO 32-bit register), but many ACCES devices do not
	have this limitation; most devices support 8, 16, or 32-bit access to any register or group thereof (as long as offsets are
	width-aligned; eg 32-bit operations require that offset % 4 == 0).  WidthFromOffset() therefore becomes complex, and
	declaratively describing each device's capabilities and restrictions is also complex.

	Another approach would be to implement a "HAL interface" (C++ calls an interface an Abstract Base Class or ABC) which the library
	would use to access the needed polymorphisms via a device-specific TDeviceHAL object it is provided ("Dependency Injection").
	Every TDeviceHAL descendant would provide not-less-than a set of device-specific constants like `highestAdcChan` from the
	introduction example.  Better would be to also include "verbs" that implement generic operations as needed for the specific
	device, like an ADC_SetRange1(iChannel, rangeSpan), but this level of interface is incredibly complex, in any generic form.

	Consider the RA1216 which expects the ADC input range to be specified as a voltage span code plus an offset in ±16-bit counts;

	This is an extreme example, an outlier; an example that forces "the most general case" to be handled ... but this is merely the
	outlier in the "ADC RangeCode" axis.  Consider the RAD242 which has a 24-bit ADC, the AD8-16 which has an 8-bit ADC, the
	USB-DIO-32I which has 1 bit per I/O Group instead of the typical 8 or 4 bits — there are *many* axes of outliers, and they all
	become necessary to handle if you try to make a truly generic library/HAL interface.

	This is why "Universal Libraries" (like NI produces) are so big and difficult to code against.

	Thus, "Protocol 2" is designed to be slightly generic, but more importantly, extensible.  Devices running old Protocol libraries
	should interoperate safely, politely, with Messages sent using new, extended, versions of the protocol.

	[
		during the writing of this I've determined the extensibility I'd designed into the protocol was insufficient to support an
		already known "requirement" for extending TDataItem lengths beyond 127 bytes, and thinking about it I realized using the
		most significant bit of a Length field as a sentinel to indicate an alternate syntax, or even one of a set, applies, does
		not provide the ability to change the LENGTH field, unless that is defined in advance at day 1.

		As a result of discussing this with Daria we've decided to just double the existing max payload lengths, in both TMessages
		and TDataItems (i.e., moving from 16 to 32 bits, and from 8 to 16 bits, respectively).  The delta overhead from this change
		limits at 25%, and *bandwidth* isn't a concern (exceeding MTU-size multiples, thus increasing TCP packet count, is of some
		concern).

		No future version of this protocol will support longer length Messages.  Instead, if an eNET- device *needs* longer
		Messages or DataItems, or needs either in *unspecified* lengths, clients will connect to a different listen_port and use a different
		protocol.  This is how Protocol 1 supports "ADC Streaming".
	]

[program overview]
	Main spawns one action-thread for handling Protocol 2.
	NOTE: Main might also spawn a singleton receive/action/send thread, or one set per "Streaming" type (ADC in the eNET-AIO case), to handle the streaming Protocol(s).
	Each Client that connects spawns a receive-thread.

	EITHER
	1	the action-thread is responsible for sending data to the correct client
	OR
	2	each per-client receive-thread spawns a send-thread and queue which is stuffed from the action-thread
	OR
	3 	a single send-thread and queue exist, created by Main and stuffed by the action-thread

	Root listen in Main run-loop receives on primary connect listen_port#; valid connections spawn receive-threads that listen on the Socket
	Multiple Clients can connect; each gets one listen-thread (and perhaps one send-queue & thread).

	[receive-threads]
		Each receive-thread passes received bytes in >= Message-sized chunks to TMessage::fromBytes to construct a TMessage instance; .fromBytes is a
		class factory method that will construct the appropriate TDataItem descendants based on the TMessage.Payload bytes' DIds.
			errors throw exceptions; the exception handler will programmatically construct a TMessage to report the detected Errors
		Regardless, the receive-thread Queues the TMessage (either the one constructed via .fromBytes() or via the caught exception handler)
			into the action-thread's Queue to be handled, and resumes waiting for additional messages.
			The TMessage that was Queued is now owned by the Action Queue;
			In the case of an error the TMessage that was received goes out of scope and is destroyed when the receive-thread run-loop loops.
	[action-thread]
		The action-thread run-loop pops a TMessage "ActionBundle" out of the Action Queue
		If the Message is a report-error message it puts it into the send-queue and the run-loop loops.
			NOTE: TMessages in the Action Queue are guaranteed to be "syntactically and semantically valid".
		Normally it constructs a basic Reply TMessage. and proceeds.
		It loops over the ActionBundle.TDataItems[] and calls .Go() on each.
			Each .Go() performs its Action, using the derived-classes' DId-specific parameters as needed, and changes its internal state into a "resultDataItem"
				Basically: once resultCode has been set (from .Go() the TDataItem .AsBytes() and .AsString() functions produce different results than before .Go(),
				as they now include the resultCode and resultValue(s) as determined during .Go().  E.g., DIO_Read1() adds 1 or 0 to indicate the input level.
			The send-thread then checks the .getResultCode(); if not ERR_SUCCESS it sets the Reply Message MId to reflect the operational error, 'E'.
			Regardless, the send-thread then adds the modified TDataItem into the Reply Message. NOTE: TDataItems in TPayloads are actually shared_ptr<TDataItem>, so the
			being-assembled Reply Message AND the ActionBundle's Message both hold a reference to the TDataItem at this point.
		Once all TDataItems[] in the ActionBundle.Payload have been executed (.Go()), and stuffed into the Reply Message, .AsBytes() is called and the byte-buffer
		produced is added to a SendBundle, which is then:
		EITHER
		1	sent to the correct Client across the socket, directly from the Action thread.
			NOTE: this means ActionBundles need to hold a reference to the Client's Socket.
		OR
		2	added to a per-Client send-thread Queue to be sent asynchronously by the per-Client send-thread.
			NOTE: this means ActionBundles need a reference to the Client's send-Queue.
		OR
		3	added to a single send-thread Queue to be sent asynchronously.
			NOTE: this means ActionBundles AND SendBundles need a reference to the Client's Socket (the SendBundle just inherits it from the ActionBundle)
		TBD: once I know more about TCP Sockets and such in Linux I can figure out which is best.

		The send-thread run-loop has now completed one loop, so the TMessage it popped out of the Action Queue goes out of scope and is destroyed.
	[1. no send-thread exists separate from the action-thread]
	[2. one send-thread per Client]
		When Main gets a Connection it spawns a listen-thread, which constructs a Send Queue then spawns a send-thread that will operate on the Send Queue.
	[3. singleton send-thread]
		Main constructs both the action-thread and send-thread; only one of each exist, each with one input Queue

	The single Action Thread serves to serialize device operations, ensuring the Actions dictated in each received Message's payload get executed "atomically",
	BUT the execution order is determined by the "parsing-finished" time, not by the "Message-received" time; i.e., each Client gets its turn in the order the
	receive-threads' constructed TMessage gets added to the Action Queue.

	The Main run-loop should act as a Watchdog, monitoring the various threads to ensure they haven't dead-locked.

	All threads (including Main) should generate log file data, with programmatically configured verbosity level;

	Logs should be retrievable via Protocol 2 Messages. (NYI)
 */

#include <algorithm>
#include <arpa/inet.h>
#include <cctype>
#include <cerrno>
#include <chrono>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <fcntl.h>
#include <filesystem>
#include <netinet/in.h>
#include <poll.h>
#include <signal.h>
#include <sys/reboot.h>
#include <sys/wait.h>
#include <unistd.h>

#include "apci.h"
#include "build_info.h"
#include "daq_state.h"
#include "socket_util.h"
#include "logging.h"
#include "TMessage.h"
#include "adc.h"
#include "config.h"
#include "DataItems/ADC_.h"
#include "DataItems/BRD_.h"
#include "DataItems/CFG_.h"
#include "DataItems/DAC_.h"
#include "DataItems/REG_.h"
#include "DataItems/SYS_.h"
#include "DataItems/TDataItem.h"
#include "aioenetd.h"
#include "WebControl/webctl_aioenetd.h"
// #define MG_ARCH MG_ARCH_NEWLIB
// extern "C" {
// #include "mongoose.h"
// }


// Cross-module daemon reset request interface.
// Implemented in aioenetd.cpp; called by DataItems/BRD_.cpp.
enum class DaemonResetRequest : int {
    None   = 0,
    Gentle = 1,
    Force  = 2
};

static volatile sig_atomic_t daemonResetRequest = static_cast<sig_atomic_t>(DaemonResetRequest::None);
static volatile sig_atomic_t forceRebootFallbackArmed = 0;

int apci = -1;
volatile sig_atomic_t done = 0;
TActionQueue ActionQueue;

static int ControlListenPort = 18767; // 0x494f, ASCII for "IO"

int AdcListenPort = ControlListenPort + 1;

pthread_t action_thread;
pthread_t controlListener_thread;
pthread_t adcListener_thread;
pthread_t webListener_thread;
static bool webListenerStarted = false;

enum class SysErrStage : __u32 { Parse = 1, Execute = 2 };

static inline __u32 ErrIndex(TError rc) { return -rc; }


// // Function to serve static files
// static void serve_static(struct mg_connection *nc, struct mg_http_message *hm) {
//     struct mg_http_serve_opts opts = { .root_dir = "/home/acces/www" };
//     mg_http_serve_dir(nc, hm, &opts);
// }

// // Function to handle API requests
// static void handle_api(struct mg_connection *nc, struct mg_http_message *hm) {
// 	if (mg_match(hm->uri, mg_str("/api/data*"), NULL) )
// 	{
// 		// Example response with dynamic data
//         mg_http_reply(nc, 200, "Content-Type: application/json\r\n", "{\"adc\": [1.23, 2.34, 3.45]}");
// 	}
// 	else
// 	{
// 		mg_http_reply(nc, 404, "Content-Type: text/plain\r\n", "Not Found!");
// 	}
// }

// // Event handler for Mongoose
// static void ev_handler(struct mg_connection *nc, int ev, void *ev_data) {
//     struct mg_http_message *hm = (struct mg_http_message *) ev_data;

//     switch (ev) {
//         case MG_EV_HTTP_MSG:
// 		    Log(std::string(hm->uri.buf, hm->uri.len));
//             if (mg_match(hm->uri, mg_str("/api/*"), NULL) ) {
//                 handle_api(nc, hm);
//             } else {
//                 serve_static(nc, hm);
//             }
//             break;
//         default:
//             break;
//     }
// }


static void SetCloseOnExec(int fd, const char *what)
{
    if (fd < 0) {
        return;
    }

    int flags = fcntl(fd, F_GETFD);
    if (flags < 0) {
        Error(std::string("fcntl(F_GETFD) failed for ")
              + what + ": " + std::strerror(errno));
        return;
    }

    if (fcntl(fd, F_SETFD, flags | FD_CLOEXEC) < 0) {
        Error(std::string("fcntl(F_SETFD, FD_CLOEXEC) failed for ")
              + what + ": " + std::strerror(errno));
    }
}

static void CloseFd(int &fd, const char *what)
{
    if (fd < 0) {
        return;
    }

    int oldFd = fd;
    fd = -1;

    if (close(oldFd) < 0) {
        Error(std::string("close(") + what + ") failed for fd "
              + std::to_string(oldFd) + ": " + std::strerror(errno));
    }
}

static std::string SystemStatusString(int status)
{
    if (status == -1) {
        return "system() failed: " + std::string(std::strerror(errno));
    }

    if (WIFEXITED(status)) {
        return "exit=" + std::to_string(WEXITSTATUS(status));
    }

    if (WIFSIGNALED(status)) {
        return "signal=" + std::to_string(WTERMSIG(status));
    }

    return "raw status=" + std::to_string(status);
}

static bool EmergencySysRqReboot()
{
    int fd = open("/proc/sysrq-trigger", O_WRONLY | O_CLOEXEC);
    if (fd < 0) {
        return false;
    }

    const char cmd = 'b'; // immediate reboot
    const ssize_t wrote = write(fd, &cmd, 1);
    close(fd);

    return wrote == 1;
}

static void ForceRebootNow(const char *reason)
{
    const char *why = reason ? reason : "unspecified reason";

    Log(std::string("FORCE reboot requested: ") + why);

    /*
        First try the civilized path.  --no-block asks systemd/init to start
        rebooting but does not require this process to wait for the transaction.
    */
    sync();

    errno = 0;
    int status = std::system(
        "PATH=/usr/sbin:/usr/bin:/sbin:/bin; "
        "systemctl --no-block reboot >/dev/null 2>&1"
    );

    if (status != 0) {
        Error("FORCE reboot: systemctl --no-block reboot did not report success: "
              + SystemStatusString(status));
    }

    /*
        Give systemd a brief chance to take over.  If it does, this process will
        usually be killed before reaching the harder fallbacks.
    */
    sleep(5);

    /*
        Harder path: ask the kernel to reboot directly.  This requires CAP_SYS_BOOT,
        which aioenetd should normally have if it is running as the privileged
        hardware daemon.
    */
    sync();

    errno = 0;
    if (reboot(RB_AUTOBOOT) != 0) {
        Error(std::string("FORCE reboot: reboot(RB_AUTOBOOT) failed: ")
              + std::strerror(errno));
    }

    /*
        Last resort: emergency SysRq reboot.  This intentionally bypasses normal
        userspace shutdown.  It belongs only in FORCE mode.
    */
    if (!EmergencySysRqReboot()) {
        Error("FORCE reboot: writing 'b' to /proc/sysrq-trigger failed");
    }

    /*
        If we somehow got here, all reboot mechanisms failed.  Do not return to
        normal daemon execution after a FORCE reset request.
    */
    _exit(127);
}

static void ArmForceRebootFallback()
{
    if (forceRebootFallbackArmed) {
        return;
    }

    forceRebootFallbackArmed = 1;

    pid_t pid = fork();
    if (pid < 0) {
        Error(std::string("BRD_Reset(FORCE): failed to fork reboot fallback: ")
              + std::strerror(errno));
        return;
    }

    if (pid == 0) {
        /*
            Child process.  Keep this deliberately simple: this process exists
            only as an out-of-band reboot hammer if the normal main()/exit_handler()
            path stalls.

            Do not Log(), malloc heavily, take locks, or call std::system() here.
            The parent is multithreaded, so the forked child should stay minimal.
        */
        setsid();

        /*
            Give the parent enough time to:
              1. send the BRD_Reset reply,
              2. leave ActionThread,
              3. run exit_handler(),
              4. reach ForceRebootNow().
        */
        sleep(12);

        /*
            No sync() here by design.  The normal path already syncs.  This fallback
            is for the case where normal shutdown is wedged, and sync() may wedge too.
        */
        (void)reboot(RB_AUTOBOOT);
        (void)EmergencySysRqReboot();

        _exit(127);
    }

    Log("BRD_Reset(FORCE): armed out-of-process reboot fallback pid "
        + std::to_string(pid));
}

void RequestDaemonReset(DaemonResetRequest request)
{
    if (request == DaemonResetRequest::Force) {
        daemonResetRequest = static_cast<sig_atomic_t>(DaemonResetRequest::Force);
        ArmForceRebootFallback();
    }
    else if (daemonResetRequest != static_cast<sig_atomic_t>(DaemonResetRequest::Force)) {
        daemonResetRequest = static_cast<sig_atomic_t>(request);
    }

    done = 1;
}

DaemonResetRequest GetDaemonResetRequest()
{
    return static_cast<DaemonResetRequest>(daemonResetRequest);
}

int main(int argc, char *argv[])
{
	Intro(argc, argv);

	InitConfig(Config);
	InitializeConfigFiles(Config);

	OpenDevFile(); // sets apci
	try
	{
		LoadConfig();
	}
	catch (const std::logic_error &e)
	{
		Error(e.what());
	};
	ApplyConfig();

	pthread_create(&action_thread, NULL, (void *(*)(void *)) & ActionThread, &ActionQueue);
	pthread_create(&controlListener_thread, NULL, ControlListenerThread, (void *)AF_INET6);
	pthread_create(&adcListener_thread, NULL, AdcListenerThread, (void *)AF_INET6);

	if (WebCtlAioEnetd_Start(&webListener_thread) == 0)
		webListenerStarted = true;
	else
		Warn("WebControl HTTP listener did not start; continuing without web dashboard");

	// struct mg_mgr mgr;
	// struct mg_connection *nc;

	// mg_mgr_init(&mgr);
	// nc = mg_http_listen(&mgr, "http://0.0.0.0:80", ev_handler, &mgr);

	// if (nc == NULL) {
	//     printf("Failed to create listener\n");
	//     return 1;
	// }

	// printf("Starting server on port 80\n");

	do
	{
		usleep(10000);
		// mg_mgr_poll(&mgr, 1000);
	} while (!done);

	// mg_mgr_free(&mgr);

	DaemonResetRequest resetRequest = GetDaemonResetRequest();

	exit_handler(0);

	if (resetRequest == DaemonResetRequest::Gentle) {
		/*
			BRD_Reset(gentle):
			exit_handler() has already run, so threads are joined, APci is closed,
			DAQ hardware was reset to safe state, and config was saved.  Now restart
			this daemon image in-place.
		*/
		execv("/proc/self/exe", argv);
		Error(std::string("BRD_Reset(gentle): execv(/proc/self/exe) failed: ")
			+ std::strerror(errno));

		execv(argv[0], argv);
		Error(std::string("BRD_Reset(gentle): execv(argv[0]) failed: ")
			+ std::strerror(errno));

		return 127;
	}
	else if (resetRequest == DaemonResetRequest::Force) {
		ForceRebootNow("BRD_Reset(FORCE)");
	}

	return 0;
}

void abort_handler(int s)
{
	done = 1;
}

void exit_handler(int s)
{
	Log("exit process starting");
	done = 1;
	AdcShutdown();
	if (DaqReady() && apci >= 0)
		apci_cancel_irq(apci, 1);

	// if (controlSocket >= 0)
	// {
	// 	shutdown(controlSocket, SHUT_RDWR);
	// 	close(controlSocket);
	// 	controlSocket = -1;
	// }
	// if (adcSocket >= 0)
	// {
	// 	shutdown(adcSocket, SHUT_RDWR);
	// 	close(adcSocket);
	// 	adcSocket = -1;
	// }
	WebCtlAioEnetd_Stop();
	ActionQueue.enqueue(nullptr);

	sleep(1);

	if (webListenerStarted)
		pthread_join(webListener_thread, NULL);
	pthread_join(adcListener_thread, NULL);
	pthread_join(controlListener_thread, NULL);
	pthread_join(action_thread, NULL);

	/* put an opened card back in the power-up state */
	if (DaqReady() && apci >= 0)
	{
		out(ofsReset, bmResetEverything);
		close(apci);
		apci = -1;
	}
	SaveConfig();

	// note __attribute__((unused)) is to silence an incorrect compiler warning
	std::time_t end_time __attribute__((unused)) = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
	Log(std::string("AIOeNET Daemon ") + BuildInfo::Version + " CLOSING, it is now: " + std::string(std::ctime(&end_time)));
}

void Intro(int argc, char **argv)
{
	// stdout is captured by systemd.  Line buffering makes each completed log
	// record promptly available to journal readers even when stdout is not a TTY.
	std::setvbuf(stdout, nullptr, _IOLBF, 0);

	// note __attribute__((unused)) is to silence an incorrect compiler warning
	const char *env = std::getenv("AIOENET_LOG_LEVEL");
	if (env)
	{
		std::string lvl = env;
		Log("AIOENET_LOG_LEVEL='" + lvl + "'");
		std::transform(lvl.begin(), lvl.end(), lvl.begin(), ::tolower);
		SetLogLevel(LogLevel::Error);
		if (lvl == "trace")
			SetLogLevel(LogLevel::Trace);
		else if (lvl == "debug")
			SetLogLevel(LogLevel::Debug);
		else if (lvl == "info")
			SetLogLevel(LogLevel::Info);
		else if (lvl == "warning" || lvl == "warn")
			SetLogLevel(LogLevel::Warning);
		else if (lvl == "error")
			SetLogLevel(LogLevel::Error);
	} else
	{
		SetLogLevel(LogLevel::Info);
		Log("AIOENET_LOG_LEVEL not set, defaulting to 'debInfoug'");
	}

	std::time_t start_time = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
	Log(std::string("AIOeNET Daemon ") + BuildInfo::Version + " STARTING, git=" + BuildInfo::GitDescribe + ", built=" + BuildInfo::BuildUtc + ", it is now: " + std::string(std::ctime(&start_time)));

	struct sigaction sigIntHandler;
	sigIntHandler.sa_handler = abort_handler;
	sigemptyset(&sigIntHandler.sa_mask);
	sigIntHandler.sa_flags = 0;

	sigaction(SIGINT, &sigIntHandler, NULL);
	sigaction(SIGABRT, &sigIntHandler, NULL);
	sigaction(SIGTERM, &sigIntHandler, NULL);

	if (argc < 2)
	{
		Trace("Warning: no tcp port specified.  Using default: " + std::to_string(ControlListenPort));
		Trace("Usage: " + std::string(argv[0]) + " {port_to_listen — (i.e., 18767)}");
	}
	else
		sscanf(argv[1], "%d", &ControlListenPort);

	AdcListenPort = ControlListenPort + 1;
	Trace("Control port: " + std::to_string(ControlListenPort));
	Trace("ADC stream port: " + std::to_string(AdcListenPort));
}

void OpenDevFile()
{
	constexpr const char *deviceDirectory = "/dev/apci";
	apci = -1;
	std::error_code ec;
	const bool exists = std::filesystem::exists(deviceDirectory, ec);
	if (ec)
	{
		SetDaqUnavailable(DaqStatus::OpenFailed, ec.value(), deviceDirectory,
			"Cannot inspect /dev/apci: " + ec.message());
		Warn("DAQ unavailable: cannot inspect /dev/apci: " + ec.message());
		return;
	}
	if (!exists)
	{
		SetDaqUnavailable(DaqStatus::DeviceDirectoryMissing, ENOENT,
			deviceDirectory, "/dev/apci is not present");
		Warn("DAQ unavailable: /dev/apci is not present; continuing in diagnostics-only mode");
		return;
	}

	std::filesystem::directory_iterator it(deviceDirectory, ec);
	const std::filesystem::directory_iterator end;
	if (ec)
	{
		SetDaqUnavailable(DaqStatus::OpenFailed, ec.value(), deviceDirectory,
			"Cannot enumerate /dev/apci: " + ec.message());
		Warn("DAQ unavailable: cannot enumerate /dev/apci: " + ec.message());
		return;
	}
	if (it == end)
	{
		SetDaqUnavailable(DaqStatus::NoDeviceFound, ENODEV, deviceDirectory,
			"/dev/apci contains no device file");
		Warn("DAQ unavailable: /dev/apci contains no device file; continuing in diagnostics-only mode");
		return;
	}

	const std::string deviceFile = it->path().string();
	apci = open(deviceFile.c_str(), O_RDONLY | O_CLOEXEC);
	if (apci < 0)
	{
		const int savedErrno = errno;
		SetDaqUnavailable(DaqStatus::OpenFailed, savedErrno, deviceFile,
			std::string("open failed: ") + std::strerror(savedErrno));
		Warn("DAQ unavailable: open(" + deviceFile + ") failed: " +
			std::strerror(savedErrno) + "; continuing in diagnostics-only mode");
		return;
	}

	SetDaqReady(deviceFile);
	Log("Opened DAQ device @ " + deviceFile);
}

void Bind(int &Socket, int &Port, void *structaddr, int iNET)
{
    struct sockaddr_in  *addr4 = static_cast<sockaddr_in *>(structaddr);
    struct sockaddr_in6 *addr6 = static_cast<sockaddr_in6 *>(structaddr);
    int result = -1;

    Socket = socket(iNET, SOCK_STREAM, 0);
    if (Socket < 0) {
        Error(std::string("socket() failed: ") + std::strerror(errno));
        exit(EXIT_FAILURE);
    }

    SetCloseOnExec(Socket, "listen socket");

    int opt = 1;
    if (setsockopt(Socket, SOL_SOCKET, SO_REUSEADDR,
                   &opt, sizeof(opt)) < 0) {
        Error(std::string("setsockopt(SO_REUSEADDR) failed: ")
              + std::strerror(errno));
        CloseFd(Socket, "listen socket after setsockopt failure");
        exit(EXIT_FAILURE);
    }

    if (iNET == AF_INET) {
        std::memset(addr4, 0, sizeof(*addr4));
        addr4->sin_family = AF_INET;
        addr4->sin_port = htons(static_cast<short>(Port));
        addr4->sin_addr.s_addr = INADDR_ANY;

        result = bind(Socket,
                      reinterpret_cast<struct sockaddr *>(addr4),
                      sizeof(sockaddr_in));
    }
    else {
        std::memset(addr6, 0, sizeof(*addr6));
        addr6->sin6_family = AF_INET6;
        addr6->sin6_port = htons(static_cast<short>(Port));
        addr6->sin6_addr = IN6ADDR_ANY_INIT;
        addr6->sin6_scope_id = 0;

        result = bind(Socket,
                      reinterpret_cast<struct sockaddr *>(addr6),
                      sizeof(sockaddr_in6));
    }

    if (result < 0) {
        int savedErrno = errno;
        Error("Bind on port " + std::to_string(Port)
              + " failed: " + std::strerror(savedErrno));

        CloseFd(Socket, "listen socket after bind failure");
        exit(EXIT_FAILURE);
    }
}

#if 0 // pre-unbind
void Bind(int &Socket, int &Port, void *structaddr, int iNET)
{
	struct sockaddr_in *addr4 = (sockaddr_in *)structaddr;
	struct sockaddr_in6 *addr6 = (sockaddr_in6 *)structaddr;
	int result = -1;

	if ((Socket = socket(iNET, SOCK_STREAM, 0)) == 0)
	{
		perror("socket failed");
		exit(EXIT_FAILURE);
	}

	int opt = 1;
	if (setsockopt(Socket, SOL_SOCKET, SO_REUSEADDR, (char *)&opt, sizeof(opt)) < 0)
	{
		Error("setsockopt failed");
		perror("setsockopt failed");
		exit(EXIT_FAILURE);
	}
	int yes = 1;
	setsockopt(Socket, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes));

	if (iNET == AF_INET)
	{
		addr4->sin_family = AF_INET;
		addr4->sin_port = htons(static_cast<short>(Port));
		addr4->sin_addr.s_addr = INADDR_ANY;
		result = bind(Socket, (struct sockaddr *)addr4, sizeof(sockaddr_in));
	}
	else // if (iNET == AF_INET6)
	{
		memset(addr6, 0, sizeof(*addr6));
		addr6->sin6_family = AF_INET6;
		addr6->sin6_port = htons(static_cast<short>(Port));
		addr6->sin6_flowinfo = 0; // 0x607ACCE5;
		addr6->sin6_addr = IN6ADDR_ANY_INIT;
		addr6->sin6_scope_id = 0;
		// addr6->sin6_scope_id = 0x0e;
		result = bind(Socket, (struct sockaddr *)addr6, sizeof(sockaddr_in6));
	}

	if (result < 0)
	{
		Error("Bind on port " + std::to_string(Port) + " failed");
		exit(EXIT_FAILURE);
	}
	// if (Port == ControlListenPort)
	// 	controlSocket = Socket;
	// if (Port == AdcListenPort)
	// 	adcSocket = Socket;
}
#endif

void Listen(int &Socket, int num)
{
	if (listen(Socket, num) < 0) // 32 connections; soft-cap or hard-cap?
	{
		Error("listen(ControlSocket) failed");
		perror("listen(ControlSocket) failed");
		exit(EXIT_FAILURE);
	}
}

void *ControlListenerThread(void *arg)
{
    const int iNET = static_cast<int>(reinterpret_cast<std::intptr_t>(arg));
    sockaddr_storage controlAddress{};
    int controlSocket = -1;
    const socklen_t controlAddressSize = sizeof(controlAddress);

    Bind(controlSocket, ControlListenPort, &controlAddress, iNET == AF_INET6 ? AF_INET6 : AF_INET);

    Trace("Listen for Control Socket");
    Listen(controlSocket, 32);
    HandleNewControlClients(controlSocket, controlAddressSize, controlAddress);
    CloseFd(controlSocket, "control listen socket");
    return nullptr;
}

struct TAdcConnectionMonitorContext
{
    int Socket = -1;
    TAdcConnectionId ConnectionId = ADC_INVALID_CONNECTION_ID;
};

static bool SocketWouldBlock(int error)
{
    if (error == EAGAIN)
        return true;
#if EWOULDBLOCK != EAGAIN
    if (error == EWOULDBLOCK)
        return true;
#endif
    return false;
}

static bool SendAdcHello(int socket, TAdcConnectionId connectionId)
{
    const __u32 helloAdc = static_cast<__u32>(connectionId) | 0x80000000u;
    const ssize_t bytesSent = SendAll(socket, &helloAdc, sizeof(helloAdc));
    if (bytesSent != static_cast<ssize_t>(sizeof(helloAdc)))
    {
        Error("TCP send of ADC Hello failed (" + std::to_string(bytesSent) + " != " + std::to_string(sizeof(helloAdc)) + ")");
        return false;
    }

    Log("Sent ADC Hello: socket=" + std::to_string(socket) + ", connection=" + std::to_string(connectionId) + ", wire=" + to_hex<__u32>(helloAdc));
    return true;
}

static void *AdcConnectionMonitorThread(void *arg)
{
    std::unique_ptr<TAdcConnectionMonitorContext> context(static_cast<TAdcConnectionMonitorContext *>(arg));
    int socket = context->Socket;
    const TAdcConnectionId connectionId = context->ConnectionId;

    pollfd descriptor{};
    descriptor.fd = socket;
    descriptor.events = POLLIN;
#ifdef POLLRDHUP
    descriptor.events |= POLLRDHUP;
#endif

    while (!done)
    {
        descriptor.revents = 0;
        const int status = poll(&descriptor, 1, 1000);
        if (status < 0)
        {
            if (errno == EINTR)
                continue;
            Error("poll() failed for ADC data connection " + std::to_string(connectionId) + ": " + std::strerror(errno));
            break;
        }
        if (status == 0)
            continue;

        short disconnectEvents = POLLERR | POLLHUP | POLLNVAL;
#ifdef POLLRDHUP
        disconnectEvents |= POLLRDHUP;
#endif
        if ((descriptor.revents & disconnectEvents) != 0)
            break;

        if ((descriptor.revents & POLLIN) != 0)
        {
            char byte = 0;
            const ssize_t received = recv(socket, &byte, sizeof(byte), MSG_PEEK | MSG_DONTWAIT);
            if (received == 0)
                break;
            if (received > 0)
            {
                Warn("Unexpected client data received on ADC streaming connection " + std::to_string(connectionId) + "; closing the connection");
                break;
            }
            if (!SocketWouldBlock(errno) && errno != EINTR)
                break;
        }
    }

    if (socket >= 0)
        shutdown(socket, SHUT_RDWR);
    AdcDataConnectionClosed(connectionId, socket);
    CloseFd(socket, "ADC data connection");
    Log("ADC data connection closed: connection=" + std::to_string(connectionId));
    return nullptr;
}

void HandleNewAdcClients(int listenSocket)
{
    Trace("Accept for ADC streaming");

    while (!done)
    {
        fd_set readfds;
        FD_ZERO(&readfds);
        FD_SET(listenSocket, &readfds);

        timeval timeout{};
        timeout.tv_sec = 1;

        const int status = select(listenSocket + 1, &readfds, nullptr, nullptr, &timeout);
        if (status < 0)
        {
            if (errno == EINTR)
                continue;
            Error("select() failed for ADC listener: " + std::string(std::strerror(errno)));
            break;
        }
        if (status == 0)
            continue;

        sockaddr_storage peerAddress{};
        socklen_t peerAddressSize = sizeof(peerAddress);
        int newSocket = accept(listenSocket, reinterpret_cast<sockaddr *>(&peerAddress), &peerAddressSize);
        if (newSocket < 0)
        {
            if (errno == EINTR)
                continue;
            Error("accept() failed for ADC connection: " + std::string(std::strerror(errno)));
            continue;
        }

        SetCloseOnExec(newSocket, "ADC data socket");

        timeval sendTimeout{};
        sendTimeout.tv_sec = 1;
        if (setsockopt(newSocket, SOL_SOCKET, SO_SNDTIMEO, &sendTimeout, sizeof(sendTimeout)) < 0)
            Warn("setsockopt(SO_SNDTIMEO) failed for ADC data socket " + std::to_string(newSocket) + ": " + std::strerror(errno));

        const TAdcConnectionId connectionId = AdcRegisterDataConnection(newSocket);
        if (connectionId == ADC_INVALID_CONNECTION_ID)
        {
            Error("Unable to allocate an ADC data connection ID");
            CloseFd(newSocket, "unregistered ADC data socket");
            continue;
        }

        if (!SendAdcHello(newSocket, connectionId))
        {
            AdcDataConnectionClosed(connectionId, newSocket);
            CloseFd(newSocket, "ADC data socket after Hello failure");
            continue;
        }

        auto *monitorContext = new TAdcConnectionMonitorContext{newSocket, connectionId};
        pthread_t monitorThread{};
        const int threadStatus = pthread_create(&monitorThread, nullptr, &AdcConnectionMonitorThread, monitorContext);
        if (threadStatus != 0)
        {
            delete monitorContext;
            Error("pthread_create(ADC connection monitor) failed: " + std::to_string(threadStatus) + ", " + std::strerror(threadStatus));
            AdcDataConnectionClosed(connectionId, newSocket);
            CloseFd(newSocket, "ADC data socket after monitor failure");
            continue;
        }

        pthread_detach(monitorThread);
    }
}

void *AdcListenerThread(void *arg)
{
    const int iNET = static_cast<int>(reinterpret_cast<std::intptr_t>(arg));
    sockaddr_storage adcAddress{};
    int adcSocket = -1;

    Bind(adcSocket, AdcListenPort, &adcAddress, iNET == AF_INET6 ? AF_INET6 : AF_INET);
    Listen(adcSocket, 32);
    HandleNewAdcClients(adcSocket);
    CloseFd(adcSocket, "ADC listen socket");
    return nullptr;
}

void SendControlHello(int socket, TAdcConnectionId connectionId)
{
	TMessageId MId_Hello = 'H';
	TPayload Payload;
	TBytes data{};
	TBytes bytes{};
	TError result;
	for (uint byt = 0; byt < sizeof(connectionId); byt++)
		data.push_back(static_cast<__u8>((connectionId >> (8 * byt)) & 0x000000FFu));
	stuff<__u16>(bytes, static_cast<__u16>(DataItemIds::TCP_ConnectionID));
	stuff<__u16>(bytes, static_cast<__u16>(data.size()));
	bytes.insert(bytes.end(), data.begin(), data.end());

	PTDataItem d2 = TDataItemBase::fromBytes(bytes, result);
	Payload.push_back(d2);
	Log("DID[TCP_ConnectionID] = ", d2->AsBytes(true));

	//__u32 dacRangeDefault = 0x3031E142; "B±10"
	//__u32 dacRangeDefault = 0x35303055;
	for (__u8 channel = 0; channel < Config.NUM_DACs; channel++)
	{
		data.clear();
		data.push_back(channel);
		for (uint byt = 0; byt < sizeof(Config.dacRanges[channel]); byt++)
			data.push_back(static_cast<__u8>((Config.dacRanges[channel] >> (8 * byt)) & 0x000000FF));
		bytes.clear();
		stuff<__u16>(bytes, static_cast<__u16>(DataItemIds::DAC_Range1));
		bytes.push_back(5);
		bytes.push_back(0);
		bytes.insert(bytes.end(), data.begin(), data.end());
		d2 = TDataItemBase::fromBytes(bytes, result);
		Payload.push_back(d2);
	}

	try
	{
		auto daqStatus = std::make_shared<TSYS_GetDaqStatus>();
		daqStatus->Go();
		Payload.push_back(daqStatus);

		auto buildInfo = std::make_shared<TSYS_GetBuildInfo>();
		buildInfo->Go();
		Payload.push_back(buildInfo);

		auto adcChans = std::make_shared<TBRD_GetNumberOfAdcChannels>();
		adcChans->Go();
		Payload.push_back(adcChans);

		if (DaqReady())
		{
			auto features = std::make_shared<TBRD_Features>();
			auto deviceID = std::make_shared<TBRD_DeviceID>();
			auto adcBaseClock = std::make_shared<TADC_BaseClock>();
			auto fpgaId = std::make_shared<TBRD_FpgaId>();
			features->Go();
			deviceID->Go();
			adcBaseClock->Go();
			fpgaId->Go();
			Payload.push_back(features);
			Payload.push_back(deviceID);
			Payload.push_back(adcBaseClock);
			Payload.push_back(fpgaId);
		}
	}
	catch (const std::exception &e)
	{
		Error(std::string("Building Control Hello failed: ") + e.what());
	}

	TMessage HelloControl = TMessage(MId_Hello, Payload);
	try
	{
		TBytes rbuf = HelloControl.AsBytes(true);
		const ssize_t bytesSent = SendAll(socket, rbuf.data(), rbuf.size());
		if (bytesSent != static_cast<ssize_t>(rbuf.size()))
		{
			Error("TCP Send of Control Hello failed (" + std::to_string(bytesSent) +
			      " != " + std::to_string(rbuf.size()) + ")");
		}
		else
		{
			Log("Sent 'Hello' to control socket " + std::to_string(socket) + ", connection " + std::to_string(connectionId) +
			    ":\n          " + HelloControl.AsString() + ", bytes=", rbuf);
		}
	}
	catch (const std::exception &e)
	{
		Error(std::string("Could not serialize Control Hello: ") + e.what());
	}
}

void Disconnect(int aClient)
{
	struct sockaddr_storage addr; // Can hold IPv4 or IPv6
	socklen_t addrSize = sizeof(addr);

	if (getpeername(aClient, reinterpret_cast<struct sockaddr *>(&addr), &addrSize) == -1)
		Error("getpeername() failed");
	else
	{
		char ipStr[INET6_ADDRSTRLEN] = {0}; // Enough space for IPv6 text
		uint16_t port = 0;

		if (addr.ss_family == AF_INET) // IPv4
		{
			auto *v4 = reinterpret_cast<struct sockaddr_in *>(&addr);
			inet_ntop(AF_INET, &(v4->sin_addr), ipStr, sizeof(ipStr));
			port = ntohs(v4->sin_port);
		}
		else if (addr.ss_family == AF_INET6) // IPv6
		{
			auto *v6 = reinterpret_cast<struct sockaddr_in6 *>(&addr);
			inet_ntop(AF_INET6, &(v6->sin6_addr), ipStr, sizeof(ipStr));
			port = ntohs(v6->sin6_port);
		}
		else
		{
			// Some other address family?
			strncpy(ipStr, "UnknownAF", sizeof(ipStr));
			port = 0;
		}

		Log("Host " + to_hex<__u32>(aClient) + " disconnected; IP: " + ipStr + ", Port " + std::to_string(port));
	}
	close(aClient);

	// // Remove from list
	// auto it = std::find(ClientList.begin(), ClientList.end(), aClient);
	// if (it != ClientList.end())
	//     ClientList.erase(it);
	// else
	//     Error("INTERNAL ERROR: Attempted to remove socket not in ClientList");
}

static bool TryParseGuardWhat(const char *what, __u32 &errIndex, __u32 &info)
{
	if (!what) return false;
	std::string s(what);

	// Expected GUARD format contains "): -<digits>" and ends with "= <hex>"
	auto p = s.find("): -");
	if (p == std::string::npos) return false;
	p += 4;

	size_t q = p;
	while (q < s.size() && std::isdigit(static_cast<unsigned char>(s[q]))) q++;
	if (q == p) return false;

	try { errIndex = static_cast<__u32>(std::stoul(s.substr(p, q - p), nullptr, 10)); }
	catch (...) { return false; }

	auto eq = s.rfind('=');
	if (eq != std::string::npos)
	{
		std::string hex = s.substr(eq + 1);
		while (!hex.empty() && std::isspace(static_cast<unsigned char>(hex.front()))) hex.erase(hex.begin());
		while (!hex.empty() && std::isspace(static_cast<unsigned char>(hex.back()))) hex.pop_back();
		if (!hex.empty() && (hex.rfind("0x", 0) == 0 || hex.rfind("0X", 0) == 0)) hex.erase(0, 2);
		try { info = static_cast<__u32>(std::stoul(hex, nullptr, 16)); } catch (...) { /* leave info as-is */ }
	}
	return true;
}

static PTDataItemBase BuildDataItem(DataItemIds did, const TBytes &payload)
{
	if (payload.size() > MaxDataItemPayload)
	{
		Error("BuildDataItem rejected payload larger than 0xFFFE bytes");
		return nullptr;
	}
	TBytes di;
	stuff<__u16>(di, static_cast<__u16>(did));
	stuff<__u16>(di, static_cast<__u16>(payload.size()));
	di.insert(di.end(), payload.begin(), payload.end());

	TError r = ERR_SUCCESS;
	try
	{
		PTDataItemBase item = TDataItemBase::fromBytes(di, r);
		if (r == ERR_SUCCESS) return item;
	}
	catch (const std::logic_error &e)
	{
		Error(std::string("BuildDataItem(") + to_hex<__u16>(static_cast<__u16>(did)) + ") threw: " + e.what());
	}
	return nullptr;
}

static PTDataItemBase BuildSysError(SysErrStage stage, __u32 errIndex, __u32 info, const std::string &text)
{
	TBytes payload;
	stuff<__u32>(payload, static_cast<__u32>(stage));
	stuff<__u32>(payload, errIndex);
	stuff<__u32>(payload, info);
	if (!text.empty()) Debug("SYS_Error detail (not placed on fixed-size wire item): " + text);
	return BuildDataItem(DataItemIds::SYS_Error, payload);
}

static PTDataItemBase BuildSysItemError(__u16 itemIndex, DataItemIds originalDid, __u32 errIndex, __u32 info, const std::string &text)
{
	TBytes payload;
	stuff<__u16>(payload, itemIndex);
	stuff<__u16>(payload, static_cast<__u16>(originalDid));
	stuff<__u32>(payload, errIndex);
	stuff<__u32>(payload, info);
	if (!text.empty()) Debug("SYS_ItemError detail (not placed on fixed-size wire item): " + text);
	return BuildDataItem(DataItemIds::SYS_ItemError, payload);
}

// bool GotMessage(char theBuffer[], int bytesRead, TMessage &parsedMessage)
// {
// 	TError result;
// 	TBytes buf(theBuffer, theBuffer + bytesRead);
// 	try
// 	{
// 		parsedMessage = TMessage::FromBytes(buf, result);
// 		Log("Received " + std::to_string(bytesRead) + " bytes on Control connection:");
// 		if (result != ERR_SUCCESS)
// 		{
// 			Error("TMessage::fromBytes(buf) returned " + std::to_string(-result) + ": " + err_msg[-result]);
// 			return false;
// 		}
// 	}
// 	catch (const std::logic_error &e)
// 	{
// 		Error("TMessage::fromBytes(buf) threw exception: " + std::string(e.what()));
// 		// todo: discard offending bytes, send 'X', continue cleanly
// 		return false;
// 	}
// 	Log("Received " + std::to_string(bytesRead) + " bytes on Control connection:\n		  " + parsedMessage.AsString());
// 	return true;
// }


void SendResponse(int Client, TMessage &aMessage);

ssize_t ReceiveFromSocket(int aSocket, std::vector<char> &buffer)
{
	char tempBuffer[65536];
	ssize_t bytesRead;
	do
	{
		bytesRead = recv(aSocket, tempBuffer, sizeof(tempBuffer), 0);
	}
	while (bytesRead < 0 && errno == EINTR && done == 0);

	if (bytesRead > 0)
	{
		buffer.insert(buffer.end(), tempBuffer, tempBuffer + bytesRead);
	}
	return bytesRead;
}

int CheckDisconnect(ssize_t bytesRead, int aSocket)
{
	if (bytesRead == 0)
		return true;
	return false;
}

bool GotMessage(const char *theBuffer, int bytesRead, TMessage &outMessage)
{
	TError result = ERR_SUCCESS;
	TBytes buf(theBuffer, theBuffer + bytesRead);

	try
	{
		outMessage = TMessage::FromBytes(buf, result);
		if (result != ERR_SUCCESS)
		{
			TMessage x('X');
			if (PTDataItemBase di = BuildSysError(SysErrStage::Parse, ErrIndex(result), 0, err_msg[-result]))
				x.addDataItem(di);
			outMessage = x;
		}
	}
	catch (const std::logic_error &e)
	{
		__u32 errIndex = ErrIndex(ERR_MSG_PARSE);
		__u32 info = 0;
		TryParseGuardWhat(e.what(), errIndex, info);

		TMessage x('X');
		if (PTDataItemBase di = BuildSysError(SysErrStage::Parse, errIndex, info, e.what())) x.addDataItem(di);
		outMessage = x;
	}
	return true;
}

void ProcessMessages(std::vector<char> &buffer, int aSocket, TAdcConnectionId connectionId)
{
	for (;;)
	{
		if (buffer.size() < (sizeof(TMessageId) + sizeof(TMessagePayloadSize))) return;

		__u32 payloadLenLe = 0;
		std::memcpy(&payloadLenLe, buffer.data() + 1, sizeof(payloadLenLe));
		__u32 payloadLen = le32toh(payloadLenLe);

		if (payloadLen > maxPayloadLength)
		{
			TMessage x('X');
			if (PTDataItemBase di = BuildSysError(SysErrStage::Parse,
				ErrIndex(ERR_MSG_LEN_MISMATCH), payloadLen,
				"Payload length exceeds maxPayloadLength"))
				x.addDataItem(di);

			// This connection cannot be safely resynchronized without accepting the
			// claimed payload.  Send the bounded error reply before shutting down
			// the socket; the receiver thread performs the single close().
			SendResponse(aSocket, x);
			buffer.clear();
			(void)shutdown(aSocket, SHUT_RDWR);
			return;
		}

		const size_t frameLen = sizeof(TMessageId) + sizeof(TMessagePayloadSize) + payloadLen + sizeof(TCheckSum);
		if (buffer.size() < frameLen) return;

		auto aMessage = std::make_shared<TMessage>();
		(void)GotMessage(buffer.data(), static_cast<int>(frameLen), *aMessage);
		auto action = new TActionQueueItem;
		action->Socket = aSocket;
		action->ConnectionId = connectionId;
		action->Message = aMessage;
		ActionQueue.enqueue(action);

		buffer.erase(buffer.begin(), buffer.begin() + frameLen);
	}
}


// void ProcessMessage(std::vector<char> &buffer, int aSocket)
// {
// 	__u32 expectedMessageLength = 0;

// 	if (buffer.size() >= 5)
// 	{
// 		expectedMessageLength = *reinterpret_cast<__u32 *>(&buffer[1]);
// 		expectedMessageLength = le32toh(expectedMessageLength);
// 		Debug("-- Expected Length: " + std::to_string(expectedMessageLength));
// 	}

// 	if (buffer.size() >= expectedMessageLength + 5 + 1)
// 	{
// 		TMessage *aMessage = new TMessage;
// 		if (GotMessage(buffer.data(), expectedMessageLength + 5 + 1, *aMessage))
// 		{
// 			TActionQueueItem *action = new TActionQueueItem{aSocket, *aMessage};
// 			ActionQueue.enqueue(action);
// 			Debug(action->theMessage.AsString(true));
// 		}
// 		buffer.erase(buffer.begin(), buffer.begin() + expectedMessageLength + 5 + 1);
// 		expectedMessageLength = 0;
// 	}
// }

struct TControlReceiverContext
{
    int Socket = -1;
    TAdcConnectionId ConnectionId = ADC_INVALID_CONNECTION_ID;
};

void *threadReceiver(void *arg)
{
    std::unique_ptr<TControlReceiverContext> context(static_cast<TControlReceiverContext *>(arg));
    const int controlSocket = context->Socket;
    const TAdcConnectionId connectionId = context->ConnectionId;
    sockaddr_storage address{};
    socklen_t addressSize = sizeof(address);

    if (getpeername(controlSocket, reinterpret_cast<sockaddr *>(&address), &addressSize) == -1)
        Error("getpeername() failed for control connection " + std::to_string(connectionId) + ": " + std::strerror(errno));
    else
    {
        char ipString[INET6_ADDRSTRLEN] = {0};
        std::uint16_t port = 0;

        if (address.ss_family == AF_INET)
        {
            const auto *ipv4 = reinterpret_cast<const sockaddr_in *>(&address);
            inet_ntop(AF_INET, &ipv4->sin_addr, ipString, sizeof(ipString));
            port = ntohs(ipv4->sin_port);
        }
        else if (address.ss_family == AF_INET6)
        {
            const auto *ipv6 = reinterpret_cast<const sockaddr_in6 *>(&address);
            inet_ntop(AF_INET6, &ipv6->sin6_addr, ipString, sizeof(ipString));
            port = ntohs(ipv6->sin6_port);
        }
        else
        {
            std::strncpy(ipString, "UnknownAF", sizeof(ipString) - 1);
        }

        Log("New control connection thread: socket=" + std::to_string(controlSocket) + ", connection=" + std::to_string(connectionId) + ", IP=" + ipString + ", port=" + std::to_string(port));
    }

    SendControlHello(controlSocket, connectionId);
    std::vector<char> buffer;

    while (!done)
    {
        const ssize_t bytesRead = ReceiveFromSocket(controlSocket, buffer);
        if (bytesRead < 0)
        {
            Error("Control recv() failed for connection " + std::to_string(connectionId) + ": " + std::to_string(errno) + ", " + std::strerror(errno));
            break;
        }
        if (CheckDisconnect(bytesRead, controlSocket))
            break;

        try
        {
            Trace("Control receiver got " + std::to_string(bytesRead) + " bytes on connection " + std::to_string(connectionId));
            ProcessMessages(buffer, controlSocket, connectionId);
        }
        catch (const std::logic_error &e)
        {
            Error(e.what());
        }
    }

    Log("Closing control receiver: socket=" + std::to_string(controlSocket) + ", connection=" + std::to_string(connectionId));
    AdcControlConnectionClosed(connectionId);
    Disconnect(controlSocket);
    return nullptr;
}

void HandleNewControlClients(int controlListenSocket, socklen_t addrSize, sockaddr_storage &addr)
{
    (void)addrSize;
    (void)addr;
    Trace("Accept for Control");

    while (!done)
    {
        fd_set readfds;
        FD_ZERO(&readfds);
        FD_SET(controlListenSocket, &readfds);

        timeval timeout{};
        timeout.tv_sec = 1;

        const int status = select(controlListenSocket + 1, &readfds, nullptr, nullptr, &timeout);
        if (status < 0)
        {
            if (errno == EINTR)
                continue;
            Error("select() failed for control listener: " + std::string(std::strerror(errno)));
            break;
        }
        if (status == 0)
            continue;

        sockaddr_storage peerAddress{};
        socklen_t peerAddressSize = sizeof(peerAddress);
        int newSocket = accept(controlListenSocket, reinterpret_cast<sockaddr *>(&peerAddress), &peerAddressSize);
        if (newSocket < 0)
        {
            if (errno == EINTR)
                continue;
            Error("accept() failed for control connection: " + std::string(std::strerror(errno)));
            continue;
        }

        SetCloseOnExec(newSocket, "control client socket");

        const TAdcConnectionId connectionId = AdcRegisterControlConnection(newSocket);
        if (connectionId == ADC_INVALID_CONNECTION_ID)
        {
            Error("Unable to allocate a control connection ID");
            CloseFd(newSocket, "unregistered control socket");
            continue;
        }

        auto *receiverContext = new TControlReceiverContext{newSocket, connectionId};
        pthread_t receiverThread{};
        const int threadStatus = pthread_create(&receiverThread, nullptr, &threadReceiver, receiverContext);
        if (threadStatus != 0)
        {
            delete receiverContext;
            Error("pthread_create(control receiver) failed: " + std::to_string(threadStatus) + ", " + std::strerror(threadStatus));
            AdcControlConnectionClosed(connectionId);
            CloseFd(newSocket, "control socket after receiver failure");
            continue;
        }

        pthread_detach(receiverThread);
        Log("New control connection: socket=" + std::to_string(newSocket) + ", connection=" + std::to_string(connectionId));
    }
}

// bool RunMessage(TMessage &aMessage)
// {
// 	bool anyError = false;
// 	Trace("RunMessage start");
// 	for (size_t i = 0; i < aMessage.DataItems.size(); i++)
// 	{
// 		Debug("RunMessage Item: " + to_hex<__u16>(__u16(i)) + " / " + to_hex<__u16>(__u16(aMessage.DataItems.size()-1)) + " = ?");// + to_hex<__u16>(__u16(item ? item->getDId())));
// 		PTDataItemBase &item = aMessage.DataItems[i];
// 		Trace("PTDataItemBase &item = aMessage.DataItems[i]; passed");
// 		DataItemIds originalDid = DataItemIds::INVALID;
// 		if (item) {
// 			originalDid = item->getDId(); // throwing Segmentation Fault, intermittently
// 		}
// 		else Debug("Item is NULL");
// 		try
// 		{
// 			if (item) item->Go();

// 			TError rc = item ? item->getResultCode() : ERR_MSG_PARSE;
// 			if (rc != ERR_SUCCESS)
// 			{
// 				anyError = true;
// 				__u32 errIndex = ErrIndex(rc);
// 				if (PTDataItemBase errItem = BuildSysItemError(static_cast<__u16>(i), originalDid, errIndex, 0, "")) item = errItem;
// 			}
// 		}
// 		catch (const std::logic_error &e)
// 		{
// 			anyError = true;
// 			__u32 errIndex = ErrIndex(ERR_MSG_PARSE);
// 			__u32 info = 0;
// 			TryParseGuardWhat(e.what(), errIndex, info);
// 			if (PTDataItemBase errItem = BuildSysItemError(static_cast<__u16>(i), originalDid, errIndex, info, e.what())) item = errItem;
// 			Error("Logic Exception in RunMessage");
// 		}
// 		catch (const std::exception &e)
// 		{
// 			anyError = true;
// 			if (PTDataItemBase errItem = BuildSysItemError(static_cast<__u16>(i), originalDid, ErrIndex(ERR_MSG_PARSE), 0, e.what())) item = errItem;
// 			Error("Exception in RunMessage");
// 		}
// 		catch (...)
// 		{
// 			anyError = true;
// 			if (PTDataItemBase errItem = BuildSysItemError(static_cast<__u16>(i), originalDid, ErrIndex(ERR_MSG_PARSE), 0, "unknown exception")) item = errItem;
// 			Error("Unknown Exception in RunMessage");
// 		}
// 	}

// 	aMessage.setMId(anyError ? 'E' : 'R');
// 	return !anyError;
// }

bool RunMessage(TMessage &aMessage)
{
	bool anyError = false;

	for (size_t i = 0; i < aMessage.DataItems.size(); ++i)
	{
		auto item = aMessage.DataItems[i];
		DataItemIds originalDid = item ? item->getDId() : DataItemIds::INVALID;

		try
		{
			if (item && item->getResultCode() == ERR_SUCCESS)
			{
				auto dictEntry = DIdDict.find(originalDid);
				if (dictEntry != DIdDict.end() && dictEntry->second.requiresDaq && !DaqReady())
				{
					const DaqStateSnapshot state = GetDaqStateSnapshot();
					item->resultCode = ERR_DAQ_UNAVAILABLE;
					item->errorInfo = static_cast<__u32>(state.lastErrno != 0 ? state.lastErrno : ENODEV);
				}
				else
				{
					item->Go();
				}
			}

			const TError rc = item ? item->getResultCode() : ERR_MSG_PARSE;
			if (rc != ERR_SUCCESS)
			{
				anyError = true;
				const __u32 info = item ? item->getErrorInfo() : 0;
				if (auto errItem = BuildSysItemError(static_cast<__u16>(i), originalDid,
					ErrIndex(rc), info, ""))
					aMessage.DataItems[i] = errItem;
			}
		}
		catch (const std::logic_error &e)
		{
			anyError = true;
			__u32 errIndex = ErrIndex(ERR_MSG_PARSE);
			__u32 info = 0;
			TryParseGuardWhat(e.what(), errIndex, info);
			if (auto errItem = BuildSysItemError(static_cast<__u16>(i), originalDid,
				errIndex, info, e.what()))
				aMessage.DataItems[i] = errItem;
			Error(std::string("DataItem logic error: ") + e.what());
		}
		catch (const std::exception &e)
		{
			anyError = true;
			if (auto errItem = BuildSysItemError(static_cast<__u16>(i), originalDid,
				ErrIndex(ERR_MSG_PARSE), 0, e.what()))
				aMessage.DataItems[i] = errItem;
			Error(std::string("DataItem exception: ") + e.what());
		}
		catch (...)
		{
			anyError = true;
			if (auto errItem = BuildSysItemError(static_cast<__u16>(i), originalDid,
				ErrIndex(ERR_MSG_PARSE), 0, "unknown exception"))
				aMessage.DataItems[i] = errItem;
			Error("Unknown DataItem exception");
		}
	}
	aMessage.setMId(anyError ? 'E' : 'R');
	return !anyError;
}


void SendResponse(int Client, TMessage &aMessage)
{
	TBytes rbuf;
	try
	{
		rbuf = aMessage.AsBytes(true);
	}
	catch (const std::length_error &e)
	{
		Error(std::string("Reply serialization exceeded protocol limits: ") + e.what());
		TMessage fallback('X');
		if (auto item = BuildSysError(SysErrStage::Execute, ErrIndex(ERR_DATAITEM_TOO_LARGE), 0, e.what()))
			fallback.addDataItem(item);
		rbuf = fallback.AsBytes(true);
	}
	catch (const std::exception &e)
	{
		Error(std::string("Reply serialization failed: ") + e.what());
		TMessage fallback('X');
		if (auto item = BuildSysError(SysErrStage::Execute, ErrIndex(ERR_MSG_PARSE), 0, e.what()))
			fallback.addDataItem(item);
		rbuf = fallback.AsBytes(true);
	}

	const ssize_t bytesSent = SendAll(Client, rbuf.data(), rbuf.size());
	if (bytesSent != static_cast<ssize_t>(rbuf.size()))
	{
		Error("TCP Send of Control reply failed (" + std::to_string(bytesSent) +
		      " != " + std::to_string(rbuf.size()) + ")");
	}
	else
	{
		Trace("sent Reply to Control Client# " + std::to_string(Client) + " " +
		      std::to_string(bytesSent) + " bytes: ", rbuf);
	}
}


int AioEnetd_RunSerialized(const char *name, std::function<int(void)> work, unsigned timeout_ms)
{
	(void)name;

	if (!work)
		return -EINVAL;

	auto donePromise = std::make_shared<std::promise<int>>();
	auto doneFuture = donePromise->get_future();

	auto action = new TActionQueueItem;
	action->Socket = -1;
	action->Work = std::move(work);
	action->Done = donePromise;

	ActionQueue.enqueue(action);

	if (doneFuture.wait_for(std::chrono::milliseconds(timeout_ms)) != std::future_status::ready)
		return -ETIMEDOUT;

	return doneFuture.get();
}

void *ActionThread(TActionQueue *Q)
{
	for (; done == 0;)
	{
		std::unique_ptr<TActionQueueItem> anAction(Q->dequeue());
		if (!anAction) continue;

		Trace("---DEQUEUED---");

		if (anAction->Work)
		{
			int workResult = 0;
			try
			{
				workResult = anAction->Work();
			}
			catch (const std::exception &e)
			{
				Error(std::string("serialized WebControl work threw exception: ") + e.what());
				workResult = -EFAULT;
			}
			catch (...)
			{
				Error("serialized WebControl work threw unknown exception");
				workResult = -EFAULT;
			}

			if (anAction->Done)
				anAction->Done->set_value(workResult);
			continue;
		}

		if (!anAction->Message)
			continue;

		if (anAction->Socket >= 0 && !AdcIsControlConnectionActive(anAction->ConnectionId, anAction->Socket))
		{
			Debug("Dropping queued work for disconnected control connection " + std::to_string(anAction->ConnectionId));
			continue;
		}

		TMessage &message = *anAction->Message;
		for (auto &item : message.DataItems)
		{
			if (item)
				item->conn = static_cast<int>(anAction->ConnectionId);
		}

		const TMessageId mid = message.getMId();
		if (mid == 'Q' || mid == 'C' || mid == 'M')
			RunMessage(message);

		if (anAction->Socket >= 0 && AdcIsControlConnectionActive(anAction->ConnectionId, anAction->Socket))
			SendResponse(anAction->Socket, message);
	}
	return nullptr;
}

// //------------------- Signal---------------------------
// #define max_BRK_attempts 3
// static void sig_handler(int sig)
// {
// 	static int sig_count = 1;
// 	Log("signal " + std::to_string(sig) + " detected " + std::to_string(sig_count++) + " time; exiting");
// 	done = true;
// 	if (sig_count > max_BRK_attempts)
// 	{
// 		Error("main runloop's `done` flag not managing to exit, Terminating!");
// 		exit(1);
// 	}
// }
// //---------------------End signal -----------------------