now lab6 eits gracefully
This commit is contained in:
@@ -4,6 +4,11 @@ CC = gcc
|
|||||||
CFLAGS = -Wall -Wextra -std=c99 -g
|
CFLAGS = -Wall -Wextra -std=c99 -g
|
||||||
LDFLAGS_MQ = -lrt
|
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
|
all: msg
|
||||||
|
|
||||||
# ===== POSIX MQ targets =====
|
# ===== POSIX MQ targets =====
|
||||||
@@ -19,7 +24,7 @@ msg_worker: worker.c common.h
|
|||||||
test_msg_server: msg
|
test_msg_server: msg
|
||||||
@echo "=== Запуск сервера POSIX MQ ==="
|
@echo "=== Запуск сервера POSIX MQ ==="
|
||||||
@echo "В другом терминале выполните: make test_msg_workers"
|
@echo "В другом терминале выполните: make test_msg_workers"
|
||||||
./msg_server 150 10 1000 800
|
./msg_server $(SERVER_ARGS)
|
||||||
|
|
||||||
test_msg_workers: msg
|
test_msg_workers: msg
|
||||||
@echo "=== Запуск нескольких пчёл ==="
|
@echo "=== Запуск нескольких пчёл ==="
|
||||||
@@ -31,12 +36,12 @@ test_msg_workers: msg
|
|||||||
# Автотест: сервер в фоне, несколько пчёл
|
# Автотест: сервер в фоне, несколько пчёл
|
||||||
test_all: msg
|
test_all: msg
|
||||||
@echo "=== Автотест POSIX MQ ==="
|
@echo "=== Автотест POSIX MQ ==="
|
||||||
./msg_server 150 10 1000 800 & \
|
./msg_server $(SERVER_ARGS) & \
|
||||||
SRV=$$!; \
|
SRV=$$!; \
|
||||||
sleep 2; \
|
sleep 2; \
|
||||||
./msg_worker 7 & \
|
./msg_worker $(WORKER_ARGS) & \
|
||||||
./msg_worker 7 & \
|
./msg_worker $(WORKER_ARGS) & \
|
||||||
./msg_worker 7 & \
|
./msg_worker $(WORKER_ARGS) & \
|
||||||
wait; \
|
wait; \
|
||||||
wait $$SRV
|
wait $$SRV
|
||||||
|
|
||||||
|
|||||||
@@ -2,16 +2,16 @@
|
|||||||
#pragma once
|
#pragma once
|
||||||
#include <sys/types.h>
|
#include <sys/types.h>
|
||||||
|
|
||||||
#define REQ_QUEUE "/winnie_req" // общая очередь запросов к серверу (POSIX name starts with '/')
|
#define REQ_QUEUE "/winnie_req"
|
||||||
#define NAME_MAXLEN 64
|
#define NAME_MAXLEN 64
|
||||||
|
|
||||||
typedef struct {
|
typedef struct {
|
||||||
pid_t pid; // PID пчелы
|
pid_t pid; // PID пчелы
|
||||||
int want; // желаемая порция
|
int want; // желаемая порция; 0 => STOP-маркер
|
||||||
char replyq[NAME_MAXLEN];// имя очереди для ответов (например, "/bee_1234")
|
char replyq[NAME_MAXLEN]; // имя очереди для ответов ("/bee_<pid>")
|
||||||
} req_msg_t;
|
} req_msg_t;
|
||||||
|
|
||||||
typedef struct {
|
typedef struct {
|
||||||
int granted; // выдано меда
|
int granted; // выдано меда; 0 => стоп
|
||||||
int remain; // остаток мёда
|
int remain; // остаток мёда
|
||||||
} rep_msg_t;
|
} rep_msg_t;
|
||||||
|
|||||||
110
lab_5/server.c
110
lab_5/server.c
@@ -1,4 +1,8 @@
|
|||||||
// server.c
|
// 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
|
#define _GNU_SOURCE
|
||||||
#include <errno.h>
|
#include <errno.h>
|
||||||
#include <mqueue.h>
|
#include <mqueue.h>
|
||||||
@@ -14,6 +18,10 @@
|
|||||||
|
|
||||||
#include "common.h"
|
#include "common.h"
|
||||||
|
|
||||||
|
#ifndef MAX_CLIENTS
|
||||||
|
#define MAX_CLIENTS 128
|
||||||
|
#endif
|
||||||
|
|
||||||
static volatile sig_atomic_t stop_flag = 0;
|
static volatile sig_atomic_t stop_flag = 0;
|
||||||
static void on_sigint(int) { stop_flag = 1; }
|
static void on_sigint(int) { stop_flag = 1; }
|
||||||
|
|
||||||
@@ -22,8 +30,40 @@ static void msleep(int ms) {
|
|||||||
nanosleep(&ts, NULL);
|
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) {
|
int main(int argc, char **argv) {
|
||||||
// Аргументы: total_honey, winnie_portion, winnie_period_ms, starvation_ms
|
|
||||||
if (argc != 5) {
|
if (argc != 5) {
|
||||||
fprintf(stderr, "Usage: %s <total_honey> <portion> <period_ms> <starvation_ms>\n", argv[0]);
|
fprintf(stderr, "Usage: %s <total_honey> <portion> <period_ms> <starvation_ms>\n", argv[0]);
|
||||||
return 2;
|
return 2;
|
||||||
@@ -37,22 +77,19 @@ int main(int argc, char **argv) {
|
|||||||
return 2;
|
return 2;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Очистим возможную старую очередь запросов
|
mq_unlink(REQ_QUEUE);
|
||||||
mq_unlink(REQ_QUEUE); // безопасно игнорировать ошибку, чтобы сбросить старые атрибуты
|
struct mq_attr attr; memset(&attr, 0, sizeof(attr));
|
||||||
struct mq_attr attr;
|
attr.mq_maxmsg = 10; // ваш системный максимум = 10
|
||||||
memset(&attr, 0, sizeof(attr));
|
attr.mq_msgsize = sizeof(req_msg_t);
|
||||||
attr.mq_maxmsg = 10; // не превышайте /proc/sys/fs/mqueue/msg_max
|
|
||||||
attr.mq_msgsize = sizeof(req_msg_t); // не превышайте /proc/sys/fs/mqueue/msgsize_max
|
|
||||||
|
|
||||||
// Создаём очередь запросов; O_RDONLY + O_CREAT, сервер читает
|
// Открываем общую очередь на O_RDWR: читаем заявки и посылаем STOP
|
||||||
mqd_t qreq = mq_open(REQ_QUEUE, O_CREAT | O_RDONLY, 0666, &attr);
|
mqd_t qreq = mq_open(REQ_QUEUE, O_CREAT | O_RDWR, 0666, &attr);
|
||||||
if (qreq == (mqd_t)-1) {
|
if (qreq == (mqd_t)-1) {
|
||||||
perror("mq_open qreq");
|
perror("mq_open qreq");
|
||||||
fprintf(stderr, "Hint: check /proc/sys/fs/mqueue/msg_max and msgsize_max\n");
|
fprintf(stderr, "Hint: ensure msg_max>=10 and msgsize_max>=%zu\n", sizeof(req_msg_t));
|
||||||
return 1;
|
return 1;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Диагностика атрибутов очереди
|
|
||||||
struct mq_attr got;
|
struct mq_attr got;
|
||||||
if (mq_getattr(qreq, &got) == 0) {
|
if (mq_getattr(qreq, &got) == 0) {
|
||||||
fprintf(stderr, "Server: q=%s maxmsg=%ld msgsize=%ld cur=%ld\n",
|
fprintf(stderr, "Server: q=%s maxmsg=%ld msgsize=%ld cur=%ld\n",
|
||||||
@@ -63,49 +100,51 @@ int main(int argc, char **argv) {
|
|||||||
fprintf(stderr, "Server: started remain=%d portion=%d period=%dms starve=%dms\n",
|
fprintf(stderr, "Server: started remain=%d portion=%d period=%dms starve=%dms\n",
|
||||||
remain, portion, period_ms, starvation_ms);
|
remain, portion, period_ms, starvation_ms);
|
||||||
|
|
||||||
// Таймер Винни‑Пуха: потребляет мёд порциями по расписанию
|
|
||||||
struct timespec now;
|
struct timespec now;
|
||||||
clock_gettime(CLOCK_MONOTONIC, &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 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;
|
long long last_feed_ns = (long long)now.tv_sec*1000000000LL + now.tv_nsec;
|
||||||
|
|
||||||
req_msg_t req;
|
req_msg_t req;
|
||||||
|
bool need_stop_broadcast = false;
|
||||||
|
|
||||||
while (!stop_flag) {
|
while (!stop_flag) {
|
||||||
clock_gettime(CLOCK_MONOTONIC, &now);
|
clock_gettime(CLOCK_MONOTONIC, &now);
|
||||||
long long now_ns = (long long)now.tv_sec*1000000000LL + now.tv_nsec;
|
long long now_ns = (long long)now.tv_sec*1000000000LL + now.tv_nsec;
|
||||||
int sleep_ms = (int)((next_eat_ns - now_ns)/1000000LL);
|
int sleep_ms = (int)((next_eat_ns - now_ns)/1000000LL);
|
||||||
if (sleep_ms < 0) sleep_ms = 0;
|
if (sleep_ms < 0) sleep_ms = 0;
|
||||||
|
|
||||||
// Ждём запрос до дедлайна следующего «приёма пищи»
|
|
||||||
struct timespec deadline = { .tv_sec = now.tv_sec + sleep_ms/1000,
|
struct timespec deadline = { .tv_sec = now.tv_sec + sleep_ms/1000,
|
||||||
.tv_nsec = now.tv_nsec + (sleep_ms%1000)*1000000L };
|
.tv_nsec = now.tv_nsec + (sleep_ms%1000)*1000000L };
|
||||||
if (deadline.tv_nsec >= 1000000000L) { deadline.tv_sec++; deadline.tv_nsec -= 1000000000L; }
|
if (deadline.tv_nsec >= 1000000000L) { deadline.tv_sec++; deadline.tv_nsec -= 1000000000L; }
|
||||||
|
|
||||||
ssize_t rd = mq_timedreceive(qreq, (char*)&req, sizeof(req), NULL, &deadline);
|
ssize_t rd = mq_timedreceive(qreq, (char*)&req, sizeof(req), NULL, &deadline);
|
||||||
if (rd >= 0) {
|
if (rd >= 0) {
|
||||||
rep_msg_t rep;
|
if (req.want != 0) {
|
||||||
if (remain > 0) {
|
/* record client's reply queue so we can notify it on shutdown */
|
||||||
int grant = req.want;
|
add_client(req.replyq);
|
||||||
if (grant > remain) grant = remain;
|
|
||||||
remain -= grant;
|
|
||||||
rep.granted = grant;
|
|
||||||
rep.remain = remain;
|
|
||||||
} else {
|
|
||||||
rep.granted = 0;
|
|
||||||
rep.remain = 0;
|
|
||||||
}
|
|
||||||
|
|
||||||
// Ответ в очередь пчелы
|
rep_msg_t rep;
|
||||||
mqd_t qrep = mq_open(req.replyq, O_WRONLY);
|
if (remain > 0) {
|
||||||
if (qrep != (mqd_t)-1) {
|
int grant = req.want;
|
||||||
mq_send(qrep, (const char*)&rep, sizeof(rep), 0);
|
if (grant > remain) grant = remain;
|
||||||
mq_close(qrep);
|
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) {
|
} else if (errno != ETIMEDOUT && errno != EAGAIN) {
|
||||||
perror("mq_timedreceive");
|
perror("mq_timedreceive");
|
||||||
}
|
}
|
||||||
|
|
||||||
// Потребление мёда по расписанию
|
|
||||||
clock_gettime(CLOCK_MONOTONIC, &now);
|
clock_gettime(CLOCK_MONOTONIC, &now);
|
||||||
now_ns = (long long)now.tv_sec*1000000000LL + now.tv_nsec;
|
now_ns = (long long)now.tv_sec*1000000000LL + now.tv_nsec;
|
||||||
if (now_ns >= next_eat_ns) {
|
if (now_ns >= next_eat_ns) {
|
||||||
@@ -118,6 +157,7 @@ int main(int argc, char **argv) {
|
|||||||
} else {
|
} else {
|
||||||
if (starvation_ms > 0 && (now_ns - last_feed_ns)/1000000LL >= starvation_ms) {
|
if (starvation_ms > 0 && (now_ns - last_feed_ns)/1000000LL >= starvation_ms) {
|
||||||
fprintf(stderr, "Winnie starved, stopping\n");
|
fprintf(stderr, "Winnie starved, stopping\n");
|
||||||
|
need_stop_broadcast = true;
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -125,11 +165,17 @@ int main(int argc, char **argv) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (remain <= 0) {
|
if (remain <= 0) {
|
||||||
msleep(100);
|
need_stop_broadcast = true;
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (need_stop_broadcast) {
|
||||||
|
fprintf(stderr, "Server: broadcasting STOP to clients\n");
|
||||||
|
send_stop_to_clients();
|
||||||
|
msleep(100);
|
||||||
|
}
|
||||||
|
|
||||||
mq_close(qreq);
|
mq_close(qreq);
|
||||||
mq_unlink(REQ_QUEUE);
|
mq_unlink(REQ_QUEUE);
|
||||||
fprintf(stderr, "Server: finished\n");
|
fprintf(stderr, "Server: finished\n");
|
||||||
|
|||||||
@@ -31,16 +31,13 @@ int main(int argc, char **argv) {
|
|||||||
char replyq[NAME_MAXLEN];
|
char replyq[NAME_MAXLEN];
|
||||||
snprintf(replyq, sizeof(replyq), "/bee_%d", (int)me);
|
snprintf(replyq, sizeof(replyq), "/bee_%d", (int)me);
|
||||||
|
|
||||||
// Создаём личную очередь ответов; сначала удалим старую
|
|
||||||
mq_unlink(replyq);
|
mq_unlink(replyq);
|
||||||
struct mq_attr attr;
|
struct mq_attr attr; memset(&attr, 0, sizeof(attr));
|
||||||
memset(&attr, 0, sizeof(attr));
|
attr.mq_maxmsg = 10; // также укладываемся в лимит
|
||||||
attr.mq_maxmsg = 8;
|
|
||||||
attr.mq_msgsize = sizeof(rep_msg_t);
|
attr.mq_msgsize = sizeof(rep_msg_t);
|
||||||
mqd_t qrep = mq_open(replyq, O_CREAT | O_RDONLY, 0666, &attr);
|
mqd_t qrep = mq_open(replyq, O_CREAT | O_RDONLY, 0666, &attr);
|
||||||
if (qrep == (mqd_t)-1) { perror("mq_open reply"); return 1; }
|
if (qrep == (mqd_t)-1) { perror("mq_open reply"); return 1; }
|
||||||
|
|
||||||
// Подключаемся к очереди запросов; возможна гонка — делаем ретраи
|
|
||||||
mqd_t qreq = (mqd_t)-1;
|
mqd_t qreq = (mqd_t)-1;
|
||||||
for (int i = 0; i < 50; i++) {
|
for (int i = 0; i < 50; i++) {
|
||||||
qreq = mq_open(REQ_QUEUE, O_WRONLY);
|
qreq = mq_open(REQ_QUEUE, O_WRONLY);
|
||||||
@@ -58,10 +55,10 @@ int main(int argc, char **argv) {
|
|||||||
unsigned seed = (unsigned)(time(NULL) ^ (uintptr_t)me);
|
unsigned seed = (unsigned)(time(NULL) ^ (uintptr_t)me);
|
||||||
|
|
||||||
while (1) {
|
while (1) {
|
||||||
int ms = 100 + (rand_r(&seed) % 600); // «дорога» 100..700 мс
|
int ms = 100 + (rand_r(&seed) % 600);
|
||||||
msleep(ms);
|
msleep(ms);
|
||||||
|
|
||||||
req_msg_t req = {0};
|
req_msg_t req; memset(&req, 0, sizeof(req));
|
||||||
req.pid = me;
|
req.pid = me;
|
||||||
req.want = portion;
|
req.want = portion;
|
||||||
strncpy(req.replyq, replyq, sizeof(req.replyq)-1);
|
strncpy(req.replyq, replyq, sizeof(req.replyq)-1);
|
||||||
@@ -78,7 +75,6 @@ int main(int argc, char **argv) {
|
|||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
if (rep.granted <= 0) {
|
if (rep.granted <= 0) {
|
||||||
// Мёд закончился
|
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
dprintf(STDOUT_FILENO, "Bee %d got %d, remain %d\n", (int)me, rep.granted, rep.remain);
|
dprintf(STDOUT_FILENO, "Bee %d got %d, remain %d\n", (int)me, rep.granted, rep.remain);
|
||||||
|
|||||||
Reference in New Issue
Block a user