#define _GNU_SOURCE #include #include #include #include #include #include #include #include #include #include #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 \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; }