| | |
| | | { |
| | | SharedMemory &shm = TestShm(); |
| | | GlobalInit(shm); |
| | | auto InitSem = [](auto id) { |
| | | auto sem_id = semget(id, 1, 0666 | IPC_CREAT); |
| | | union semun init_val; |
| | | init_val.val = 1; |
| | | semctl(sem_id, 0, SETVAL, init_val); |
| | | return; |
| | | }; |
| | | |
| | | MQId id = ShmMsgQueue::NewId(); |
| | | InitSem(id); |
| | | MQId server_id = ShmMsgQueue::NewId(); |
| | | ShmMsgQueue server(server_id, shm, 1000); |
| | | |
| | | const int timeout = 1000; |
| | | const uint32_t data_size = 1001; |
| | |
| | | std::string str(data_size, 'a'); |
| | | auto Writer = [&](int writer_id, uint64_t n) { |
| | | MQId cli_id = ShmMsgQueue::NewId(); |
| | | InitSem(cli_id); |
| | | |
| | | ShmMsgQueue mq(cli_id, shm, 64); |
| | | MsgI msg; |
| | |
| | | |
| | | for (uint64_t i = 0; i < n; ++i) { |
| | | msg.AddRef(); |
| | | while (!mq.TrySend(id, msg.Offset())) {} |
| | | while (!mq.TrySend({server.Id(), server.AbsAddr()}, msg.Offset())) {} |
| | | ++nwrite; |
| | | } |
| | | }; |
| | | auto Reader = [&](int reader_id, std::atomic<bool> *run, bool isfork) { |
| | | ShmMsgQueue mq(id, shm, 1000); |
| | | ShmMsgQueue &mq = server; |
| | | auto now = []() { return steady_clock::now(); }; |
| | | auto tm = now(); |
| | | while (*run) { |
| | |
| | | req_body.set_topic("topic"); |
| | | req_body.set_data(msg_content); |
| | | auto req_head(InitMsgHead(GetType(req_body), client_proc_id, cli.id())); |
| | | req_head.add_route()->set_mq_id(cli.id()); |
| | | return cli.Send(srv.id(), req_head, req_body); |
| | | auto route = req_head.add_route(); |
| | | route->set_mq_id(cli.id()); |
| | | route->set_abs_addr(cli.AbsAddr()); |
| | | return cli.Send({srv.id(), srv.AbsAddr()}, req_head, req_body); |
| | | }; |
| | | |
| | | Req(); |
| | |
| | | DEFER1(req.Release()); |
| | | |
| | | if (req.ParseHead(req_head) && req_head.type() == kMsgTypeRequestTopic) { |
| | | auto src_id = req_head.route()[0].mq_id(); |
| | | MQInfo src_mq = {req_head.route()[0].mq_id(), req_head.route()[0].abs_addr()}; |
| | | auto Reply = [&]() { |
| | | MsgRequestTopic reply_body; |
| | | reply_body.set_topic("topic"); |
| | | reply_body.set_data(msg_content); |
| | | auto reply_head(InitMsgHead(GetType(reply_body), server_proc_id, srv.id(), req_head.msg_id())); |
| | | return srv.Send(src_id, reply_head, reply_body); |
| | | return srv.Send(src_mq, reply_head, reply_body); |
| | | }; |
| | | Reply(); |
| | | } |