Skip to content

Commit f0e2bb8

Browse files
catchashu10catchashu10
committed
Added test cases for Performance
1 parent fabca1d commit f0e2bb8

5 files changed

Lines changed: 256 additions & 0 deletions

File tree

‎CMakeLists.txt‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@ include_directories("${CMAKE_CURRENT_SOURCE_DIR}/common")
3838
include_directories("${CMAKE_CURRENT_BINARY_DIR}/producer")
3939
include_directories("${CMAKE_CURRENT_BINARY_DIR}/consumer")
4040
include_directories("${CMAKE_CURRENT_BINARY_DIR}/sys_admin")
41+
include_directories("${CMAKE_CURRENT_BINARY_DIR}/test")
4142

4243
# dmq_grpc_proto library
4344
add_library(dmq_grpc_proto
@@ -149,5 +150,15 @@ target_link_libraries(sys_admin_client
149150
sys_admin
150151
)
151152

153+
# Test executable
154+
add_executable( performace
155+
test/performace.cc
156+
)
157+
target_link_libraries (performace
158+
producer
159+
consumer_group
160+
-static-libstdc++ -static-libgcc
161+
162+
)
152163
# Set compiler flags for position-independent code for building shared libraries
153164
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fPIC")

‎test/consumer.h‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
#ifndef CONSUMER_H
2+
#define CONSUMER_H
3+
4+
#include <memory>
5+
#include <string>
6+
#include <vector>
7+
8+
struct MessageResponse {
9+
std::string key;
10+
std::string value;
11+
std::string topic;
12+
int timestamp;
13+
};
14+
15+
class Consumer {
16+
private:
17+
class Impl;
18+
std::unique_ptr<Impl> impl_;
19+
std::string consumer_id;
20+
21+
public:
22+
Consumer(const std::vector<std::string> &bootstrap_servers, std::string consumer_id);
23+
~Consumer();
24+
std::vector<MessageResponse> ConsumeMessage(std::string group_id, std::string topic, int partition, int offset, int max_messages);
25+
std::string get_consumer_id();
26+
};
27+
28+
#endif // CONSUMER_H

‎test/consumer_group.h‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
#ifndef CONSUMER_GROUP_H
2+
#define CONSUMER_GROUP_H
3+
4+
#include <memory>
5+
#include <string>
6+
#include <vector>
7+
#include <unordered_map>
8+
#include "consumer.h"
9+
10+
struct topic_state {
11+
std::string topic;
12+
int partition;
13+
int offset;
14+
};
15+
16+
class ConsumerGroup {
17+
private:
18+
std::string tag;
19+
std::string group_id;
20+
std::vector<std::unique_ptr<Consumer>> consumers_;
21+
std::unordered_map<std::string,
22+
std::unordered_map<std::string, std::vector<topic_state>>> consumer_topic_state_;
23+
std::unordered_map<std::string, std::string> topic_partition_consumer_;
24+
25+
bool IsTopicConsumed(std::string topic, int partition);
26+
27+
public:
28+
ConsumerGroup(std::string tag, std::string group_id);
29+
~ConsumerGroup();
30+
bool AddConsumer(const std::vector<std::string>& bootstrap_servers, std::string consumer_id, std::vector<std::string> topics, std::vector<int> partitions, std::vector<int> offsets);
31+
bool RemoveConsumer(std::string consumer_id);
32+
std::vector<MessageResponse> ConsumeMessage(std::string topic, int partition, int max_messages);
33+
void PrintConsumerGroup();
34+
};
35+
36+
#endif // CONSUMER_GROUP_H

‎test/performace.cc‎

Lines changed: 158 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,158 @@
1+
#include <iostream>
2+
#include <vector>
3+
#include <string>
4+
#include <random>
5+
#include <sys/types.h>
6+
#include <sys/wait.h>
7+
#include <unistd.h>
8+
#include "producer.h"
9+
#include "consumer_group.h"
10+
#include "chrono"
11+
12+
using namespace std;
13+
14+
// Utility function to generate random messages
15+
string GenerateRandomString(size_t length) {
16+
static const string charset = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890";
17+
random_device rd;
18+
mt19937 gen(rd());
19+
uniform_int_distribution<> dis(0, charset.size() - 1);
20+
string result;
21+
for (size_t i = 0; i < length; ++i) {
22+
result += charset[dis(gen)];
23+
}
24+
return result;
25+
}
26+
27+
// Test case 1: Single Producer, Single Consumer Group
28+
void TestCase1() {
29+
cout << "Running TestCase1: Single Producer, Single Consumer Group" << endl;
30+
31+
vector<string> bootstrap_servers = {"localhost:8080"};
32+
Producer producer(bootstrap_servers, 1000, 500, "producer1");
33+
ConsumerGroup consumerGroup("tag1", "group1");
34+
35+
consumerGroup.AddConsumer(bootstrap_servers, "consumer1", {"test_topic"}, {0}, {0});
36+
37+
const int totalMessages = 10000;
38+
auto startTime = chrono::high_resolution_clock::now();
39+
40+
// Producer sends messages
41+
for (int i = 0; i < totalMessages; ++i) {
42+
producer.ProduceMessage("key" + to_string(i), GenerateRandomString(20), "test_topic");
43+
}
44+
45+
// Consumer fetches messages
46+
vector<MessageResponse> responses = consumerGroup.ConsumeMessage("test_topic", 0, totalMessages);
47+
48+
auto endTime = chrono::high_resolution_clock::now();
49+
chrono::duration<double> elapsed = endTime - startTime;
50+
51+
cout << "TestCase1 Results: " << endl;
52+
cout << "Total messages: " << totalMessages << endl;
53+
cout << "Elapsed time: " << elapsed.count() << " seconds" << endl;
54+
cout << "Throughput: " << totalMessages / elapsed.count() << " messages/second" << endl;
55+
}
56+
57+
// Test case 2: Multiple Producers, Single Consumer Group (Using Processes)
58+
void TestCase2WithProcesses() {
59+
cout << "Running TestCase2: Multiple Producers, Single Consumer Group (Using Processes)" << endl;
60+
61+
vector<string> bootstrap_servers = {"localhost:8080"};
62+
ConsumerGroup consumerGroup("tag2", "group2");
63+
consumerGroup.AddConsumer(bootstrap_servers, "consumer1", {"test_topic"}, {0}, {0});
64+
65+
const int messagesPerProducer = 5000;
66+
const int numProducers = 3;
67+
68+
// Fork processes for producers
69+
for (int i = 0; i < numProducers; ++i) {
70+
pid_t pid = fork();
71+
if (pid == 0) { // Child process
72+
Producer producer(bootstrap_servers, 1000, 500, "producer" + to_string(i));
73+
for (int j = 0; j < messagesPerProducer; ++j) {
74+
producer.ProduceMessage("key" + to_string(i) + "_" + to_string(j),
75+
GenerateRandomString(20), "test_topic");
76+
}
77+
exit(0); // Exit the child process
78+
} else if (pid < 0) {
79+
cerr << "Error forking process for producer " << i << endl;
80+
exit(1);
81+
}
82+
}
83+
84+
// Wait for all child processes to finish
85+
for (int i = 0; i < numProducers; ++i) {
86+
int status;
87+
wait(&status);
88+
}
89+
90+
// Consumer fetches messages
91+
vector<MessageResponse> responses = consumerGroup.ConsumeMessage("test_topic", 0, numProducers * messagesPerProducer);
92+
93+
cout << "TestCase2 Results: " << endl;
94+
cout << "Total messages produced: " << numProducers * messagesPerProducer << endl;
95+
cout << "Messages consumed: " << responses.size() << endl;
96+
}
97+
98+
// Test case 3: High Volume Stress Test (Using Processes)
99+
void TestCase3WithProcesses() {
100+
cout << "Running TestCase3: High Volume Stress Test (Using Processes)" << endl;
101+
102+
vector<string> bootstrap_servers = {"localhost:8080"};
103+
vector<Producer> producers;
104+
ConsumerGroup consumerGroup("tag3", "group3");
105+
106+
// Adding consumers
107+
for (int i = 0; i < 5; ++i) {
108+
consumerGroup.AddConsumer(bootstrap_servers, "consumer" + to_string(i), {"test_topic"}, {i}, {0});
109+
}
110+
111+
const int messagesPerProducer = 100000;
112+
const int numProducers = 10;
113+
114+
// Fork processes for producers
115+
for (int i = 0; i < numProducers; ++i) {
116+
pid_t pid = fork();
117+
if (pid == 0) { // Child process
118+
Producer producer(bootstrap_servers, 1000, 500, "producer" + to_string(i));
119+
for (int j = 0; j < messagesPerProducer; ++j) {
120+
producer.ProduceMessage("key" + to_string(i) + "_" + to_string(j),
121+
GenerateRandomString(20), "test_topic");
122+
}
123+
exit(0); // Exit the child process
124+
} else if (pid < 0) {
125+
cerr << "Error forking process for producer " << i << endl;
126+
exit(1);
127+
}
128+
}
129+
130+
// Wait for all child processes to finish
131+
for (int i = 0; i < numProducers; ++i) {
132+
int status;
133+
wait(&status);
134+
}
135+
136+
cout << "All producers completed." << endl;
137+
138+
// Simulate consumers fetching messages
139+
auto startTime = chrono::high_resolution_clock::now();
140+
141+
vector<MessageResponse> responses = consumerGroup.ConsumeMessage("test_topic", 0, numProducers * messagesPerProducer);
142+
143+
auto endTime = chrono::high_resolution_clock::now();
144+
chrono::duration<double> elapsed = endTime - startTime;
145+
146+
cout << "TestCase3 Results: " << endl;
147+
cout << "Total messages consumed: " << responses.size() << endl;
148+
cout << "Elapsed time: " << elapsed.count() << " seconds" << endl;
149+
cout << "Throughput: " << (numProducers * messagesPerProducer) / elapsed.count() << " messages/second" << endl;
150+
}
151+
152+
// Main function to execute all test cases
153+
int main() {
154+
TestCase1();
155+
TestCase2WithProcesses();
156+
TestCase3WithProcesses();
157+
return 0;
158+
}

‎test/producer.h‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
#ifndef MESSAGE_QUEUE_PRODUCER_H
2+
#define MESSAGE_QUEUE_PRODUCER_H
3+
4+
#include <string>
5+
#include <vector>
6+
#include <memory>
7+
8+
class Producer {
9+
public:
10+
// Constructor to initialize producer with bootstrap servers
11+
Producer(const std::vector<std::string>& bootstrap_servers, int flush_threshold, int flush_interval_ms, const std::string& producer_id);
12+
13+
~Producer(); // Declare the destructor
14+
15+
// Produces a message to the message queue
16+
bool ProduceMessage(const std::string& key, const std::string& value, const std::string& topic);
17+
18+
private:
19+
class Impl; // Forward declaration of the implementation class
20+
std::unique_ptr<Impl> impl_; // Pointer to the implementation class
21+
};
22+
23+
#endif // MESSAGE_QUEUE_PRODUCER_H

0 commit comments

Comments
 (0)