now lab6 eits gracefully

lab_5/Makefile

@@ -0,0 +1,62 @@
+# Makefile for lab 10.5 - POSIX message queues
+
+CC = gcc
+CFLAGS = -Wall -Wextra -std=c99 -g
+LDFLAGS_MQ = -lrt
+
+# SERVER_ARGS: <total_honey> <portion> <period_ms> <starvation_ms>
+# WORKER_ARGS: <honey_portion>
+SERVER_ARGS = 1000 15 500 500
+WORKER_ARGS = 7
+
+all: msg
+
+# ===== POSIX MQ targets =====
+msg: msg_server msg_worker
+
+msg_server: server.c common.h
+	$(CC) $(CFLAGS) -o $@ $< $(LDFLAGS_MQ)
+
+msg_worker: worker.c common.h
+	$(CC) $(CFLAGS) -o $@ $< $(LDFLAGS_MQ)
+
+# ===== Tests =====
+test_msg_server: msg
+	@echo "=== Запуск сервера POSIX MQ ==="
+	@echo "В другом терминале выполните: make test_msg_workers"
+	./msg_server $(SERVER_ARGS)
+
+test_msg_workers: msg
+	@echo "=== Запуск нескольких пчёл ==="
+	./msg_worker 7 & \
+	./msg_worker 9 & \
+	./msg_worker 5 & \
+	wait
+
+# Автотест: сервер в фоне, несколько пчёл
+test_all: msg
+	@echo "=== Автотест POSIX MQ ==="
+	./msg_server $(SERVER_ARGS) & \
+	SRV=$$!; \
+	sleep 2; \
+	./msg_worker $(WORKER_ARGS) & \
+	./msg_worker $(WORKER_ARGS) & \
+	./msg_worker $(WORKER_ARGS) & \
+	wait; \
+	wait $$SRV
+
+# Очистка
+clean:
+	@echo "Очистка..."
+	rm -f msg_server msg_worker
+
+help:
+	@echo "Available targets:"
+	@echo "  msg              - Build POSIX message queue programs"
+	@echo "  test_msg_server  - Run MQ server (use workers in another terminal)"
+	@echo "  test_msg_workers - Run several worker processes"
+	@echo "  test_all         - Automatic end-to-end test"
+	@echo "  clean            - Remove built files"
+	@echo "  help             - Show this help"
+
+.PHONY: all msg test_msg_server test_msg_workers test_all clean help

lab_5/common.h

@@ -0,0 +1,17 @@
+// common.h
+#pragma once
+#include <sys/types.h>
+
+#define REQ_QUEUE   "/winnie_req"
+#define NAME_MAXLEN 64
+
+typedef struct {
+    pid_t pid;                 // PID пчелы
+    int   want;                // желаемая порция; 0 => STOP-маркер
+    char  replyq[NAME_MAXLEN]; // имя очереди для ответов ("/bee_<pid>")
+} req_msg_t;
+
+typedef struct {
+    int granted;  // выдано меда; 0 => стоп
+    int remain;   // остаток мёда
+} rep_msg_t;

lab_5/server.c

@@ -0,0 +1,183 @@
+// C
+// lab_5/server.c (modified)
+// - collect callers' replyq names on requests
+// - send a rep_msg_t {granted=0, remain=0} to each client on shutdown
+
+#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 <unistd.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;
+}

lab_5/worker.c

@@ -0,0 +1,87 @@
+// worker.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 <unistd.h>
+
+#include "common.h"
+
+static void msleep(int ms) {
+    struct timespec ts = { .tv_sec = ms/1000, .tv_nsec = (ms%1000)*1000000L };
+    nanosleep(&ts, NULL);
+}
+
+int main(int argc, char **argv) {
+    if (argc != 2) {
+        fprintf(stderr, "Usage: %s <bee_portion>\n", argv[0]);
+        return 2;
+    }
+    int portion = atoi(argv[1]);
+    if (portion <= 0) { fprintf(stderr, "portion must be >0\n"); return 2; }
+
+    pid_t me = getpid();
+    char replyq[NAME_MAXLEN];
+    snprintf(replyq, sizeof(replyq), "/bee_%d", (int)me);
+
+    mq_unlink(replyq);
+    struct mq_attr attr; memset(&attr, 0, sizeof(attr));
+    attr.mq_maxmsg = 10;                       // также укладываемся в лимит
+    attr.mq_msgsize = sizeof(rep_msg_t);
+    mqd_t qrep = mq_open(replyq, O_CREAT | O_RDONLY, 0666, &attr);
+    if (qrep == (mqd_t)-1) { perror("mq_open reply"); return 1; }
+
+    mqd_t qreq = (mqd_t)-1;
+    for (int i = 0; i < 50; i++) {
+        qreq = mq_open(REQ_QUEUE, O_WRONLY);
+        if (qreq != (mqd_t)-1) break;
+        if (errno != ENOENT) { perror("mq_open req"); break; }
+        msleep(100);
+    }
+    if (qreq == (mqd_t)-1) {
+        perror("mq_open req");
+        mq_close(qrep);
+        mq_unlink(replyq);
+        return 1;
+    }
+
+    unsigned seed = (unsigned)(time(NULL) ^ (uintptr_t)me);
+
+    while (1) {
+        int ms = 100 + (rand_r(&seed) % 600);
+        msleep(ms);
+
+        req_msg_t req; memset(&req, 0, sizeof(req));
+        req.pid = me;
+        req.want = portion;
+        strncpy(req.replyq, replyq, sizeof(req.replyq)-1);
+
+        if (mq_send(qreq, (const char*)&req, sizeof(req), 0) == -1) {
+            perror("mq_send");
+            break;
+        }
+
+        rep_msg_t rep;
+        ssize_t rd = mq_receive(qrep, (char*)&rep, sizeof(rep), NULL);
+        if (rd == -1) {
+            perror("mq_receive");
+            break;
+        }
+        if (rep.granted <= 0) {
+            break;
+        }
+        dprintf(STDOUT_FILENO, "Bee %d got %d, remain %d\n", (int)me, rep.granted, rep.remain);
+    }
+
+    mq_close(qreq);
+    mq_close(qrep);
+    mq_unlink(replyq);
+    return 0;
+}