/* Részlet a listener eseménykezelőjéből; Linux API. */
int fd = accept4(listener_fd, NULL, NULL,
                 SOCK_NONBLOCK | SOCK_CLOEXEC);

#ifndef DIAG_TCP_H
#define DIAG_TCP_H

#include <stdbool.h>
#include <stddef.h>

#define DIAG_IN_CAP  4096u
#define DIAG_OUT_CAP 1024u

enum diag_core_result { DIAG_MORE, DIAG_REPLY, DIAG_BAD };
enum diag_step {
    DIAG_KEEP, DIAG_DONE, DIAG_BAD_FRAME,
    DIAG_SYS_ERROR, DIAG_CORE_ERROR
};

typedef enum diag_core_result (*diag_reply_fn)(
    const unsigned char *input, size_t input_size,
    size_t *consumed,
    unsigned char *output, size_t output_capacity,
    size_t *produced);

struct diag_io {
    int fd;
    bool draining;
    size_t used;
    size_t off, end;
    unsigned char input[DIAG_IN_CAP];
    unsigned char output[DIAG_OUT_CAP];
};

/* A core-adapter (src/tcp_reply.c): egy kérés = egy rekordsor. */
enum diag_core_result diag_core_reply(
    const unsigned char *input, size_t input_size,
    size_t *consumed,
    unsigned char *output, size_t output_capacity,
    size_t *produced);

enum diag_step diag_read(struct diag_io *, diag_reply_fn);
enum diag_step diag_write(struct diag_io *, diag_reply_fn);
enum diag_step diag_quiesce(struct diag_io *);

#endif

#include "diag_tcp.h"

#include <errno.h>
#include <string.h>
#include <sys/socket.h>

static enum diag_step prepare(struct diag_io *c,
                              diag_reply_fn reply)
{
    if (c->draining)
        return DIAG_DONE;
    if (c->used == 0)
        return DIAG_KEEP;

    size_t consumed = 0;
    size_t produced = 0;
    enum diag_core_result r = reply(
        c->input, c->used, &consumed,
        c->output, sizeof c->output, &produced);

    if (r == DIAG_BAD)
        return DIAG_BAD_FRAME;

    if (r == DIAG_MORE) {
        if (consumed != 0 || produced != 0)
            return DIAG_CORE_ERROR;
        return c->used == sizeof c->input
             ? DIAG_BAD_FRAME : DIAG_KEEP;
    }

    if (r != DIAG_REPLY ||
        consumed == 0 || consumed > c->used ||
        produced == 0 || produced > sizeof c->output)
        return DIAG_CORE_ERROR;

    /*
     * A reply() már visszatért: nincs élő borrowed
     * parsernézet, ezért a bemenet tömöríthető.
     */
    memmove(c->input, c->input + consumed,
            c->used - consumed);
    c->used -= consumed;
    c->off = 0;
    c->end = produced;
    return DIAG_KEEP;
}

enum diag_step diag_read(struct diag_io *c,
                         diag_reply_fn reply)
{
    if (c->draining || c->off != c->end)
        return DIAG_CORE_ERROR;
    if (c->used == sizeof c->input)
        return DIAG_BAD_FRAME;

    ssize_t n = recv(c->fd, c->input + c->used,
                     sizeof c->input - c->used, 0);

    if (n > 0) {
        c->used += (size_t)n;
        return prepare(c, reply);
    }
    if (n == 0)
        return c->used == 0 ? DIAG_DONE : DIAG_BAD_FRAME;
    if (errno == EINTR || errno == EAGAIN ||
        errno == EWOULDBLOCK)
        return DIAG_KEEP;
    return DIAG_SYS_ERROR;
}

enum diag_step diag_write(struct diag_io *c,
                          diag_reply_fn reply)
{
    if (c->off >= c->end)
        return DIAG_CORE_ERROR;

    ssize_t n = send(c->fd, c->output + c->off,
                     c->end - c->off, MSG_NOSIGNAL);

    if (n < 0) {
        if (errno == EINTR || errno == EAGAIN ||
            errno == EWOULDBLOCK)
            return DIAG_KEEP;
        return DIAG_SYS_ERROR;
    }
    if (n == 0) {
        errno = EIO;
        return DIAG_SYS_ERROR;
    }

    c->off += (size_t)n;
    if (c->off != c->end)
        return DIAG_KEEP;

    c->off = c->end = 0;
    return prepare(c, reply);
}

enum diag_step diag_quiesce(struct diag_io *c)
{
    c->draining = true;
    c->used = 0;  /* A még el nem fogadott kérések eldobása. */
    return c->off == c->end ? DIAG_DONE : DIAG_KEEP;
}

gcc -std=c17 -D_GNU_SOURCE -Wall -Wextra -Wpedantic \
  -Wconversion -Wshadow -O2 -g -Iinclude \
  -c src/tcp_io.c -o tcp_io.gcc.o

clang -std=c17 -D_GNU_SOURCE -Wall -Wextra -Wpedantic \
  -Wconversion -Wshadow -O2 -g -Iinclude \
  -c src/tcp_io.c -o tcp_io.clang.o

#include "diag.h"
#include "diag_tcp.h"

#include <stdio.h>
#include <string.h>

/*
 * Egy TCP-kérés pontosan egy LF-fel lezárt rekordsor. A válasz az
 * a sor, amelyet a CLI erre az egyetlen sorból álló bemenetre a
 * stdout-ra (OK) vagy a stderr-re (ERR) írna.
 */
enum diag_core_result diag_core_reply(
    const unsigned char *input, size_t input_size,
    size_t *consumed,
    unsigned char *output, size_t output_capacity,
    size_t *produced)
{
    const unsigned char *lf = memchr(input, '\n', input_size);
    if (lf == NULL)
        return DIAG_MORE;   /* a felső korlátot a tcp_io.c tartja */

    size_t line = (size_t)(lf - input) + 1;
    dg_parser *parser = dg_create(NULL);
    int n;

    if (parser == NULL) {
        n = snprintf((char *)output, output_capacity,
                     "ERR code=alloc\n");
    } else if (dg_feed(parser, (dg_view){ input, line }) &&
               dg_finish(parser)) {
        dg_report r = dg_get_report(parser);
        n = snprintf((char *)output, output_capacity,
                     "OK records=%zu sum=%llu bytes=%zu\n",
                     r.records, r.sum, r.bytes);
    } else {
        dg_error e = dg_get_error(parser);
        n = snprintf((char *)output, output_capacity,
                     "ERR code=%s offset=%zu\n",
                     dg_code_name(e.code), e.offset);
    }
    dg_destroy(parser);   /* abort: lezárás finish nélkül is */

    if (n < 0 || (size_t)n >= output_capacity)
        return DIAG_BAD;
    *consumed = line;
    *produced = (size_t)n;
    return DIAG_REPLY;
}

/* Linkpróba: core + TCP I/O + core-adapter, valódi socketpáron. */
#include "diag_tcp.h"

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>

#define CHECK(condition) do {                                      \
    if (!(condition)) {                                            \
        (void)fprintf(stderr, "CHECK failed at line %d\n", __LINE__); \
        exit(EXIT_FAILURE);                                        \
    }                                                              \
} while (0)

int main(void)
{
    static const char request[] = "a=7\ncpu=12\nA=1\nmem=";
    static const char expected[] =
        "OK records=1 sum=7 bytes=4\n"
        "OK records=1 sum=12 bytes=7\n"
        "ERR code=syntax offset=0\n";
    int sv[2];
    char reply[256];
    size_t got = 0;

    CHECK(socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0);
    CHECK(write(sv[1], request, sizeof request - 1)
          == (ssize_t)(sizeof request - 1));
    CHECK(shutdown(sv[1], SHUT_WR) == 0);

    struct diag_io c = { .fd = sv[0] };
    enum diag_step step = diag_read(&c, diag_core_reply);

    while (step == DIAG_KEEP) {
        step = c.off != c.end ? diag_write(&c, diag_core_reply)
                              : diag_read(&c, diag_core_reply);
    }
    /* A félbemaradt "mem=" kérés EOF-nál keretszintű hiba. */
    CHECK(step == DIAG_BAD_FRAME);

    for (;;) {
        ssize_t n = recv(sv[1], reply + got, sizeof reply - got,
                         MSG_DONTWAIT);
        if (n <= 0)
            break;
        got += (size_t)n;
    }
    CHECK(got == sizeof expected - 1);
    CHECK(memcmp(reply, expected, got) == 0);

    close(sv[0]);
    close(sv[1]);
    return 0;
}

gcc -std=c17 -O2 -g -Wall -Wextra -Wpedantic -Wconversion -Wshadow \
    -Werror -D_GNU_SOURCE -Iinclude \
    src/tcp_io.c src/tcp_reply.c tests/test_tcp_link.c \
    build/gcc/libdiag.a -o build/gcc/test_tcp_link
build/gcc/test_tcp_link

/* Dispatch-részlet; a speciális listener/signal tokenek
   feldolgozása ezt megelőzi. */
uint64_t token = events[k].data.u64;
uint32_t index = (uint32_t)token;
uint32_t generation = (uint32_t)(token >> 32);

if (index >= CONNECTION_LIMIT)
    continue;
if (!slots[index].active ||
    slots[index].generation != generation)
    continue;

/* src/diag_server.c -- referencia-szerver: listener, epoll, határidők,
 * signalfd-alapú graceful shutdown. ISO C17 + Linux API. */
#include "diag_tcp.h"

#include <arpa/inet.h>
#include <errno.h>
#include <netinet/in.h>
#include <signal.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include <sys/signalfd.h>
#include <sys/socket.h>
#include <time.h>
#include <unistd.h>

/* A 17.1 határidői; tesztbuildben -D kapcsolóval rövidíthetők. */
#ifndef DIAG_IDLE_MS
#define DIAG_IDLE_MS 30000
#endif
#ifndef DIAG_FRAME_MS
#define DIAG_FRAME_MS 5000
#endif
#ifndef DIAG_REPLY_MS
#define DIAG_REPLY_MS 5000
#endif
#ifndef DIAG_SHUTDOWN_MS
#define DIAG_SHUTDOWN_MS 3000
#endif

#define CONNECTION_LIMIT 64
#define ACCEPT_BURST 16
#define RETRY_MS 100
#define TOKEN_LISTENER UINT64_MAX
#define TOKEN_SIGNAL (UINT64_MAX - 1)

enum phase { PH_IDLE, PH_FRAME, PH_REPLY, PH_DRAIN_PEER };

struct slot {
    struct diag_io io;
    uint32_t generation;
    bool active;
    bool retired;
    enum phase phase;
    int64_t deadline;             /* monoton óra, ms */
};

static struct slot slots[CONNECTION_LIMIT];
static size_t active_count;
static int ep = -1;
static int listener = -1;
static bool listening;            /* a listener az epollban van-e */
static int64_t listener_retry = -1;
static bool stopping;
static int64_t stop_deadline;

static int64_t now_ms(void)
{
    struct timespec ts;
    (void)clock_gettime(CLOCK_MONOTONIC, &ts);
    return (int64_t)ts.tv_sec * 1000 + ts.tv_nsec / 1000000;
}

static uint64_t token_of(size_t i)
{
    return (uint64_t)slots[i].generation << 32 | (uint64_t)i;
}

static int set_interest(size_t i, int op)
{
    struct slot *s = &slots[i];
    struct epoll_event ev = { .data.u64 = token_of(i) };

    if (s->phase != PH_DRAIN_PEER && s->io.off < s->io.end)
        ev.events = EPOLLOUT;
    else
        ev.events = EPOLLIN | EPOLLRDHUP;
    return epoll_ctl(ep, op, s->io.fd, &ev);
}

static void resume_listener(void)
{
    struct epoll_event ev = { .events = EPOLLIN,
                              .data.u64 = TOKEN_LISTENER };

    if (stopping || listening || listener < 0)
        return;
    if (epoll_ctl(ep, EPOLL_CTL_ADD, listener, &ev) == 0) {
        listening = true;
        listener_retry = -1;
    }
}

static void pause_listener(int64_t retry_at)
{
    if (listening) {
        (void)epoll_ctl(ep, EPOLL_CTL_DEL, listener, NULL);
        listening = false;
    }
    listener_retry = retry_at;    /* -1: szabad rekeszre várunk */
}

static void teardown(size_t i)
{
    struct slot *s = &slots[i];
    int fd = s->io.fd;

    s->active = false;            /* 1. logikailag inaktív */
    (void)epoll_ctl(ep, EPOLL_CTL_DEL, fd, NULL);
    s->io.fd = -1;
    if (close(fd) == -1)          /* egyszer; hiba esetén sem újra */
        perror("close");
    if (s->generation == UINT32_MAX)
        s->retired = true;        /* körbefordulás helyett nyugdíj */
    --active_count;
    resume_listener();
}

static void refresh(size_t i, bool new_reply)
{
    struct slot *s = &slots[i];
    enum phase p;

    if (s->io.off < s->io.end)
        p = PH_REPLY;
    else if (s->io.used > 0)
        p = PH_FRAME;
    else
        p = PH_IDLE;

    /* Fázisváltáskor indul új határidő; bájtonként nem hosszabbodik. */
    if (p != s->phase || (p == PH_REPLY && new_reply)) {
        int64_t span = p == PH_REPLY ? DIAG_REPLY_MS
                     : p == PH_FRAME ? DIAG_FRAME_MS : DIAG_IDLE_MS;
        s->phase = p;
        s->deadline = now_ms() + span;
    }
}

static void begin_drain_peer(size_t i)
{
    struct slot *s = &slots[i];

    if (shutdown(s->io.fd, SHUT_WR) == -1) {
        teardown(i);
        return;
    }
    s->phase = PH_DRAIN_PEER;
    s->deadline = stop_deadline;
    if (set_interest(i, EPOLL_CTL_MOD) == -1)
        teardown(i);
}

static void handle_step(size_t i, enum diag_step step, bool new_reply)
{
    switch (step) {
    case DIAG_KEEP:
        refresh(i, new_reply);
        if (set_interest(i, EPOLL_CTL_MOD) == -1)
            teardown(i);
        return;
    case DIAG_DONE:
        if (slots[i].io.draining)
            begin_drain_peer(i);  /* leállítás: a kimenet kiürült */
        else
            teardown(i);          /* rendezett EOF, nincs függő válasz */
        return;
    case DIAG_CORE_ERROR:
        (void)fprintf(stderr, "diag_server: core contract violation\n");
        teardown(i);
        return;
    case DIAG_BAD_FRAME:
    case DIAG_SYS_ERROR:
        teardown(i);
        return;
    }
}

static void drain_peer(size_t i)
{
    unsigned char sink[1024];
    ssize_t n = recv(slots[i].io.fd, sink, sizeof sink, 0);

    if (n > 0)
        return;                   /* eldobjuk; a szint-trigger újra jelez */
    if (n == -1 && (errno == EINTR || errno == EAGAIN ||
                    errno == EWOULDBLOCK))
        return;
    teardown(i);                  /* EOF vagy hiba */
}

static void accept_burst(void)
{
    for (int k = 0; k < ACCEPT_BURST; ++k) {
        size_t i = 0;
        while (i < CONNECTION_LIMIT && (slots[i].active || slots[i].retired))
            ++i;
        if (i == CONNECTION_LIMIT) {
            pause_listener(-1);
            return;
        }

        int fd = accept4(listener, NULL, NULL,
                         SOCK_NONBLOCK | SOCK_CLOEXEC);
        if (fd == -1) {
            if (errno == EINTR || errno == ECONNABORTED)
                continue;
            if (errno == EMFILE || errno == ENFILE ||
                errno == ENOBUFS || errno == ENOMEM)
                pause_listener(now_ms() + RETRY_MS);
            else if (errno != EAGAIN && errno != EWOULDBLOCK)
                perror("accept4");
            return;
        }

        struct slot *s = &slots[i];
        s->io = (struct diag_io){ .fd = fd };
        ++s->generation;
        s->phase = PH_IDLE;
        s->deadline = now_ms() + DIAG_IDLE_MS;
        if (set_interest(i, EPOLL_CTL_ADD) == -1) {
            s->io.fd = -1;
            (void)close(fd);      /* még az accept ág a tulajdonos */
            continue;
        }
        s->active = true;         /* tulajdonátadás a rekesznek */
        ++active_count;
    }
}

static void start_stop(void)
{
    if (stopping)
        return;
    stopping = true;
    stop_deadline = now_ms() + DIAG_SHUTDOWN_MS;

    pause_listener(-1);
    if (close(listener) == -1)
        perror("close");
    listener = -1;

    for (size_t i = 0; i < CONNECTION_LIMIT; ++i) {
        if (!slots[i].active || slots[i].phase == PH_DRAIN_PEER)
            continue;
        if (diag_quiesce(&slots[i].io) == DIAG_DONE)
            begin_drain_peer(i);
        else
            handle_step(i, DIAG_KEEP, false);
    }
}

static void dispatch(const struct epoll_event *ev, int sfd)
{
    uint64_t token = ev->data.u64;

    if (token == TOKEN_LISTENER) {
        accept_burst();
        return;
    }
    if (token == TOKEN_SIGNAL) {
        struct signalfd_siginfo info;
        while (read(sfd, &info, sizeof info) == (ssize_t)sizeof info)
            start_stop();
        return;
    }

    uint32_t i = (uint32_t)token;
    uint32_t generation = (uint32_t)(token >> 32);
    if (i >= CONNECTION_LIMIT || !slots[i].active ||
        slots[i].generation != generation)
        return;                   /* elavult esemény */

    struct slot *s = &slots[i];
    if (s->phase == PH_DRAIN_PEER) {
        drain_peer(i);
    } else if (ev->events & EPOLLERR) {
        teardown(i);
    } else if (s->io.off < s->io.end) {
        size_t used = s->io.used;
        enum diag_step step = diag_write(&s->io, diag_core_reply);
        handle_step(i, step, s->io.used < used);
    } else {
        handle_step(i, diag_read(&s->io, diag_core_reply), true);
    }
}

static void expire(int64_t now)
{
    if (listener_retry >= 0 && now >= listener_retry)
        resume_listener();

    for (size_t i = 0; i < CONNECTION_LIMIT; ++i) {
        if (slots[i].active &&
            (now >= slots[i].deadline || (stopping && now >= stop_deadline)))
            teardown(i);
    }
}

static int next_timeout(void)
{
    int64_t next = -1;

    if (listener_retry >= 0)
        next = listener_retry;
    if (stopping && (next < 0 || stop_deadline < next))
        next = stop_deadline;
    for (size_t i = 0; i < CONNECTION_LIMIT; ++i) {
        if (slots[i].active && (next < 0 || slots[i].deadline < next))
            next = slots[i].deadline;
    }
    if (next < 0)
        return -1;

    int64_t wait = next - now_ms();
    return wait <= 0 ? 0 : wait > 60000 ? 60000 : (int)wait;
}

static int setup(unsigned long port, int *sfd)
{
    sigset_t mask;
    struct sockaddr_in addr = {
        .sin_family = AF_INET,
        .sin_port = htons((uint16_t)port),
        .sin_addr.s_addr = htonl(INADDR_LOOPBACK),
    };
    socklen_t length = sizeof addr;
    int one = 1;

    if (sigemptyset(&mask) == -1 || sigaddset(&mask, SIGTERM) == -1 ||
        sigaddset(&mask, SIGINT) == -1 ||
        sigprocmask(SIG_BLOCK, &mask, NULL) == -1)
        return -1;
    *sfd = signalfd(-1, &mask, SFD_NONBLOCK | SFD_CLOEXEC);
    ep = epoll_create1(EPOLL_CLOEXEC);
    listener = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
    if (*sfd == -1 || ep == -1 || listener == -1)
        return -1;

    struct epoll_event ev = { .events = EPOLLIN, .data.u64 = TOKEN_SIGNAL };
    if (setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one) ||
        bind(listener, (struct sockaddr *)&addr, sizeof addr) == -1 ||
        listen(listener, 128) == -1 ||
        getsockname(listener, (struct sockaddr *)&addr, &length) == -1 ||
        epoll_ctl(ep, EPOLL_CTL_ADD, *sfd, &ev) == -1)
        return -1;

    resume_listener();
    if (!listening)
        return -1;
    if (printf("LISTEN 127.0.0.1:%u\n", (unsigned)ntohs(addr.sin_port)) < 0 ||
        fflush(stdout) == EOF)
        return -1;
    return 0;
}

int main(int argc, char **argv)
{
    unsigned long port = 0;
    int sfd = -1;
    int status = 0;

    if (argc > 2) {
        (void)fprintf(stderr, "ERR code=usage\n");
        return 2;
    }
    if (argc == 2) {
        char *end;
        errno = 0;
        port = strtoul(argv[1], &end, 10);
        if (errno != 0 || end == argv[1] || *end != '\0' || port > 65535) {
            (void)fprintf(stderr, "ERR code=usage\n");
            return 2;
        }
    }

    if (setup(port, &sfd) == -1) {
        perror("diag_server: setup");
        status = 5;
    } else {
        struct epoll_event events[CONNECTION_LIMIT + 2];

        while (!stopping || active_count > 0) {
            int n = epoll_wait(ep, events, CONNECTION_LIMIT + 2,
                               next_timeout());
            if (n == -1) {
                if (errno == EINTR)
                    continue;
                perror("epoll_wait");
                status = 5;
                break;
            }
            for (int k = 0; k < n; ++k)
                dispatch(&events[k], sfd);
            expire(now_ms());     /* minden köteg után, nem csak timeoutkor */
        }
    }

    for (size_t i = 0; i < CONNECTION_LIMIT; ++i) {
        if (slots[i].active)
            teardown(i);
    }
    if (listener >= 0)
        (void)close(listener);
    if (sfd >= 0)
        (void)close(sfd);
    if (ep >= 0)
        (void)close(ep);
    return status;
}

gcc -std=c17 -O2 -g -Wall -Wextra -Wpedantic -Wconversion -Wshadow \
    -Werror -D_GNU_SOURCE -Iinclude \
    src/diag_server.c src/tcp_io.c src/tcp_reply.c src/diag.c \
    -o build/gcc/diag_server

gcc -std=c17 -O1 -g -D_GNU_SOURCE -Iinclude \
    -DDIAG_IDLE_MS=1500 -DDIAG_FRAME_MS=500 \
    -DDIAG_REPLY_MS=500 -DDIAG_SHUTDOWN_MS=1000 \
    -fsanitize=address,undefined -fno-omit-frame-pointer \
    src/diag_server.c src/tcp_io.c src/tcp_reply.c src/diag.c \
    -o build/gcc/diag_server_test
python3 tests/server_test.py build/gcc/diag_server_test

#!/usr/bin/env python3
"""tests/server_test.py SERVER -- a referencia-szerver elfogadási tesztje.

A szervert rövidített határidőkkel fordítva kell átadni:
IDLE=1500, FRAME=500, REPLY=500, SHUTDOWN=1000 ms.
"""
import signal
import socket
import subprocess
import sys
import threading
import time

IDLE, FRAME, SHUTDOWN = 1.5, 0.5, 1.0
LIMIT = 64


def start(binary):
    proc = subprocess.Popen([binary], stdout=subprocess.PIPE, text=True)
    line = proc.stdout.readline().strip()
    assert line.startswith("LISTEN 127.0.0.1:"), line
    return proc, int(line.rsplit(":", 1)[1])


def connect(port):
    s = socket.create_connection(("127.0.0.1", port), timeout=3)
    return s


def recv_until_closed(s, timeout=3):
    s.settimeout(timeout)
    data = b""
    while True:
        chunk = s.recv(4096)
        if not chunk:
            return data
        data += chunk


def recv_lines(s, count, timeout=3):
    s.settimeout(timeout)
    data = b""
    while data.count(b"\n") < count:
        chunk = s.recv(4096)
        assert chunk, "a kapcsolat idő előtt bezárult"
        data += chunk
    return data


def closed_within(s, seconds):
    """Igaz, ha a szerver a megadott időn belül lezárja a kapcsolatot."""
    s.settimeout(seconds)
    try:
        while s.recv(4096):
            pass
        return True
    except (socket.timeout, TimeoutError):
        return False
    except ConnectionResetError:
        return True


def test_pipelined_and_error(port):
    s = connect(port)
    s.sendall(b"a=7\ncpu=12\nA=1\n")
    assert recv_lines(s, 3) == (b"OK records=1 sum=7 bytes=4\n"
                                b"OK records=1 sum=12 bytes=7\n"
                                b"ERR code=syntax offset=0\n")
    s.close()


def test_partial_input(port):
    s = connect(port)
    for b in b"cpu=12\n":                # bájtonként, szünetekkel
        s.sendall(bytes([b]))
        time.sleep(0.02)
    assert recv_lines(s, 1) == b"OK records=1 sum=12 bytes=7\n"
    s.close()


def test_max_request_splits(port):
    request = b"abcdefghijklmnop=65535\n"    # 23 bájt, a leghosszabb
    s = connect(port)
    s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
    for split in range(len(request) + 1):
        s.sendall(request[:split])
        time.sleep(0.005)
        s.sendall(request[split:])
        assert recv_lines(s, 1) == b"OK records=1 sum=65535 bytes=23\n"
    s.close()


def test_eof(port):
    s = connect(port)
    s.sendall(b"a=1\n")
    s.shutdown(socket.SHUT_WR)            # rendezett EOF teljes kérés után
    assert recv_until_closed(s) == b"OK records=1 sum=1 bytes=4\n"
    s.close()

    s = connect(port)
    s.sendall(b"mem=")
    s.shutdown(socket.SHUT_WR)            # félbemaradt kérés: nincs válasz
    assert recv_until_closed(s) == b""
    s.close()


def test_timeouts(port):
    s = connect(port)
    s.sendall(b"a=")                      # befejezetlen kérés
    t0 = time.monotonic()
    assert closed_within(s, FRAME + 1.0)
    assert time.monotonic() - t0 >= FRAME - 0.05
    s.close()

    s = connect(port)                     # tétlen kapcsolat
    t0 = time.monotonic()
    assert closed_within(s, IDLE + 1.0)
    assert time.monotonic() - t0 >= IDLE - 0.05
    s.close()

    s = connect(port)                     # csöpögtetés nem hosszabbít
    t0 = time.monotonic()
    s.sendall(b"a")
    closed = False
    for _ in range(40):
        time.sleep(0.05)
        try:
            s.sendall(b"a")
        except (BrokenPipeError, ConnectionResetError):
            closed = True
            break
    assert closed or closed_within(s, 0.5)
    assert time.monotonic() - t0 < FRAME + 1.0
    s.close()


def test_connection_limit(port):
    conns = [connect(port) for _ in range(LIMIT)]
    for c in conns:                       # mind a 64 kiszolgált
        c.sendall(b"a=1\n")
        assert recv_lines(c, 1) == b"OK records=1 sum=1 bytes=4\n"

    extra = connect(port)                 # a kernel elfogadja, mi nem
    extra.sendall(b"b=2\n")
    extra.settimeout(0.3)
    try:
        assert extra.recv(64) == b"", "a 65. kapcsolat nem kaphat választ"
    except (socket.timeout, TimeoutError):
        pass

    conns.pop().close()                   # egy rekesz felszabadul
    assert recv_lines(extra, 1) == b"OK records=1 sum=2 bytes=4\n"
    extra.close()
    for c in conns:
        c.close()


def test_backpressure(port):
    count = 50000
    s = connect(port)
    replies = []

    def reader():
        replies.append(recv_lines(s, count, timeout=10))

    t = threading.Thread(target=reader)
    t.start()
    s.sendall(b"".join(b"a=%d\n" % (i % 65536) for i in range(count)))
    t.join()
    expected = b"".join(b"OK records=1 sum=%d bytes=%d\n"
                        % (i % 65536, len(b"a=%d\n" % (i % 65536)))
                        for i in range(count))
    assert replies[0] == expected, "elveszett, duplikált vagy sérült válasz"
    s.close()


def test_graceful_shutdown(binary):
    proc, port = start(binary)
    done = connect(port)                  # kész válasz, majd tétlen
    done.sendall(b"a=1\n")
    assert recv_lines(done, 1) == b"OK records=1 sum=1 bytes=4\n"
    partial = connect(port)               # félbemaradt kérés
    partial.sendall(b"b=")
    flood = connect(port)                 # küld, de nem olvas
    flood.setblocking(False)
    stop = threading.Event()

    def flooder():
        while not stop.is_set():
            try:
                flood.send(b"c=1\n" * 1024)
            except BlockingIOError:
                time.sleep(0.01)
            except OSError:
                return

    t = threading.Thread(target=flooder)
    t.start()
    time.sleep(0.3)

    t0 = time.monotonic()
    proc.send_signal(signal.SIGTERM)
    assert recv_until_closed(done) == b""        # féllezárás, nincs új adat
    assert recv_until_closed(partial) == b""     # el nem fogadott kérés
    rc = proc.wait(timeout=SHUTDOWN + 2)
    elapsed = time.monotonic() - t0
    stop.set()
    t.join()
    assert rc == 0, rc
    assert elapsed < SHUTDOWN + 1.0, elapsed     # a flood sem tartja fel
    for c in (done, partial, flood):
        c.close()
    try:
        connect(port).close()
        raise AssertionError("leállás után új kapcsolat")
    except ConnectionRefusedError:
        pass


def main():
    binary = sys.argv[1]
    proc, port = start(binary)
    try:
        for test in (test_pipelined_and_error, test_partial_input,
                     test_max_request_splits, test_eof,
                     test_timeouts, test_connection_limit, test_backpressure):
            test(port)
            print("ok", test.__name__)
    finally:
        proc.send_signal(signal.SIGTERM)
        rc = proc.wait(timeout=5)
    assert rc == 0, rc
    test_graceful_shutdown(binary)
    print("ok test_graceful_shutdown")


if __name__ == "__main__":
    main()

#include "diag_tcp.h"
#include <stdint.h>
#include <sys/epoll.h>

static uint32_t active_interest(const struct diag_io *c)
{
    return c->off < c->end
         ? EPOLLOUT
         : EPOLLIN | EPOLLRDHUP;
}