CS-LABS/lab_5/server.c

#define _GNU_SOURCE
#include <errno.h>
#include <mqueue.h>
#include <signal.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <time.h>

#include "common.h"

#ifndef MAX_CLIENTS
#define MAX_CLIENTS 128
#endif

static volatile sig_atomic_t stop_flag = 0;
static void on_sigint(int) { stop_flag = 1; }

static void msleep(int ms) {
    struct timespec ts = {.tv_sec = ms / 1000, .tv_nsec = (ms % 1000) * 1000000L};
    nanosleep(&ts, NULL);
}

/* Simple client registry (stores unique reply queue names) */
static char clients[MAX_CLIENTS][NAME_MAXLEN];
static int client_count = 0;

static void add_client(const char *name) {
    if (!name || name[0] == '\0') return;
    for (int i = 0; i < client_count; ++i) {
        if (strncmp(clients[i], name, NAME_MAXLEN) == 0) return;
    }
    if (client_count < MAX_CLIENTS) {
        strncpy(clients[client_count], name, NAME_MAXLEN - 1);
        clients[client_count][NAME_MAXLEN - 1] = '\0';
        client_count++;
    }
}

/* Send a STOP reply (granted=0) to all recorded clients */
static void send_stop_to_clients(void) {
    rep_msg_t stoprep;
    memset(&stoprep, 0, sizeof(stoprep));
    stoprep.granted = 0;
    stoprep.remain = 0;
    for (int i = 0; i < client_count; ++i) {
        mqd_t q = mq_open(clients[i], O_WRONLY);
        if (q == (mqd_t) -1) {
            // best-effort: skip if cannot open
            continue;
        }
        mq_send(q, (const char *) &stoprep, sizeof(stoprep), 0);
        mq_close(q);
    }
}

int main(int argc, char **argv) {
    if (argc != 5) {
        fprintf(stderr, "Usage: %s <total_honey> <portion> <period_ms> <starvation_ms>\n", argv[0]);
        return 2;
    }
    int remain = atoi(argv[1]);
    int portion = atoi(argv[2]);
    int period_ms = atoi(argv[3]);
    int starvation_ms = atoi(argv[4]);
    if (remain < 0 || portion <= 0 || period_ms <= 0 || starvation_ms < 0) {
        fprintf(stderr, "Bad args\n");
        return 2;
    }

    mq_unlink(REQ_QUEUE);
    struct mq_attr attr;
    memset(&attr, 0, sizeof(attr));
    attr.mq_maxmsg = 10; // ваш системный максимум = 10
    attr.mq_msgsize = sizeof(req_msg_t);

    // Открываем общую очередь на O_RDWR: читаем заявки и посылаем STOP
    mqd_t qreq = mq_open(REQ_QUEUE, O_CREAT | O_RDWR, 0666, &attr);
    if (qreq == (mqd_t) -1) {
        perror("mq_open qreq");
        fprintf(stderr, "Hint: ensure msg_max>=10 and msgsize_max>=%zu\n", sizeof(req_msg_t));
        return 1;
    }

    struct mq_attr got;
    if (mq_getattr(qreq, &got) == 0) {
        fprintf(stderr, "Server: q=%s maxmsg=%ld msgsize=%ld cur=%ld\n",
                REQ_QUEUE, got.mq_maxmsg, got.mq_msgsize, got.mq_curmsgs);
    }

    signal(SIGINT, on_sigint);
    fprintf(stderr, "Server: started remain=%d portion=%d period=%dms starve=%dms\n",
            remain, portion, period_ms, starvation_ms);

    struct timespec now;
    clock_gettime(CLOCK_MONOTONIC, &now);
    long long next_eat_ns = (long long) now.tv_sec * 1000000000LL + now.tv_nsec + (long long) period_ms * 1000000LL;
    long long last_feed_ns = (long long) now.tv_sec * 1000000000LL + now.tv_nsec;

    req_msg_t req;
    bool need_stop_broadcast = false;

    while (!stop_flag) {
        clock_gettime(CLOCK_MONOTONIC, &now);
        long long now_ns = (long long) now.tv_sec * 1000000000LL + now.tv_nsec;
        int sleep_ms = (int) ((next_eat_ns - now_ns) / 1000000LL);
        if (sleep_ms < 0) sleep_ms = 0;

        struct timespec deadline = {
            .tv_sec = now.tv_sec + sleep_ms / 1000,
            .tv_nsec = now.tv_nsec + (sleep_ms % 1000) * 1000000L
        };
        if (deadline.tv_nsec >= 1000000000L) {
            deadline.tv_sec++;
            deadline.tv_nsec -= 1000000000L;
        }

        ssize_t rd = mq_timedreceive(qreq, (char *) &req, sizeof(req), NULL, &deadline);
        if (rd >= 0) {
            if (req.want != 0) {
                /* record client's reply queue so we can notify it on shutdown */
                add_client(req.replyq);

                rep_msg_t rep;
                if (remain > 0) {
                    int grant = req.want;
                    if (grant > remain) grant = remain;
                    remain -= grant;
                    rep.granted = grant;
                    rep.remain = remain;
                } else {
                    rep.granted = 0;
                    rep.remain = 0;
                }
                mqd_t qrep = mq_open(req.replyq, O_WRONLY);
                if (qrep != (mqd_t) -1) {
                    mq_send(qrep, (const char *) &rep, sizeof(rep), 0);
                    mq_close(qrep);
                }
            }
        } else if (errno != ETIMEDOUT && errno != EAGAIN) {
            perror("mq_timedreceive");
        }

        clock_gettime(CLOCK_MONOTONIC, &now);
        now_ns = (long long) now.tv_sec * 1000000000LL + now.tv_nsec;
        if (now_ns >= next_eat_ns) {
            if (remain > 0) {
                int eat = portion;
                if (eat > remain) eat = remain;
                remain -= eat;
                last_feed_ns = now_ns;
                fprintf(stderr, "Winnie eats %d, remain=%d\n", eat, remain);
            } else {
                if (starvation_ms > 0 && (now_ns - last_feed_ns) / 1000000LL >= starvation_ms) {
                    fprintf(stderr, "Winnie starved, stopping\n");
                    need_stop_broadcast = true;
                    break;
                }
            }
            next_eat_ns = now_ns + (long long) period_ms * 1000000LL;
        }

        if (remain <= 0) {
            need_stop_broadcast = true;
            break;
        }
    }

    if (need_stop_broadcast) {
        fprintf(stderr, "Server: broadcasting STOP to clients\n");
        send_stop_to_clients();
        msleep(100);
    }

    mq_close(qreq);
    mq_unlink(REQ_QUEUE);
    fprintf(stderr, "Server: finished\n");
    return 0;
}