tests.t2 source code

1 /+ dub.sdl:
2     name "t2"
3     dflags "-I../source"
4     lflags "-lcares"
5     dependency "hio" version="*"
6     debugVersions "hioepoll"
7     debugVersions "hiosocket"
8 +/
9 
10 module tests.t2;
11 
12 import std.experimental.logger;
13 import std.datetime;
14 import std.string;
15 import std.algorithm;
16 import std.range;
17 
18 import core.atomic;
19 
20 import hio.socket;
21 import hio.scheduler;
22 
23 shared int ops;
24 
25 enum client_tasks = 16;
26 enum clients = 2;
27 enum servers = 2;
28 enum duration = 15.seconds;
29 
30 void client_task()
31 {
32     while(true)
33     {
34         auto s = new HioSocket();
35         scope(exit)
36         {
37             s.close();
38         }
39         try
40         {
41             s.connect("127.0.0.1:12345", 1.seconds);
42             if (s.connected)
43             {
44                 s.send("hello".representation, 1.seconds);
45             }
46             s.close();
47             hlSleep(100.hnsecs);
48         }
49         catch(LoopShutdownException)
50         {
51             info("got shutdown");
52             return;
53         }
54         catch(Exception e)
55         {
56             errorf("%s", e);
57             return;
58         }
59         tracef("next iteration");
60     }
61 }
62 
63 void client_thread()
64 {
65     auto tasks = iota(client_tasks).map!(i => task(&client_task)).array;
66     tasks.each!(t => t.start());
67     tasks.each!(t => t.wait(2*duration));
68 }
69 
70 void handler(HioSocket s)
71 {
72     scope(exit)
73     {
74         s.close();
75     }
76     try
77     {
78         auto message = s.recv(16, 200.msecs);
79         if (!message.error && !message.timedout)
80         {
81             assert(message.input == "hello".representation, "msg: %s, sock: %s".format(message, s));
82         }
83     }
84     catch(LoopShutdownException)
85     {
86         return;
87     }
88     catch(Exception e)
89     {
90         error("server task exception: %s", s);
91     }
92 }
93 
94 void server(int so, int n)
95 {
96     auto sock = new HioSocket(so);
97     scope(exit)
98     {
99         sock.close();
100     }
101     while(true)
102     {
103         try
104         {
105             auto client_socket = sock.accept();
106             ops.atomicOp!"+="(1);
107             task(&handler, client_socket).start;
108         }
109         catch(LoopShutdownException)
110         {
111             return;
112         }
113         catch(Exception e)
114         {
115             errorf("server exception: %s", e);
116             return;
117         }
118     }
119 }
120 
121 
122 void main()
123 {
124     globalLogLevel = LogLevel.info;
125     App({
126         auto server_socket = new HioSocket();
127         server_socket.bind("0.0.0.0:12345");
128         server_socket.listen(2048);
129 
130         auto server_threads = iota(servers).map!(i => threaded(&server, server_socket.fileno, i).start).array;
131 
132         auto client_threads = iota(clients).map!(i => threaded(&client_thread).start).array;
133 
134         hlSleep(duration);
135         //globalLogLevel = LogLevel.trace;
136         client_threads.each!(t => t.shutdownThreadLoop());
137         server_threads.each!(t => t.shutdownThreadLoop());
138 
139         client_threads.each!(t => t.wait());
140         server_threads.each!(t => t.wait());
141 
142         server_socket.close();
143         infof("done %d", ops);
144     });
145 }